services/keepproxy/keepproxy.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         "errors"
   9         "flag"
  10         "fmt"
  11         "io"
  12         "io/ioutil"
  13         "log"
  14         "net"
  15         "net/http"
  16         "os"
  17         "os/signal"
  18         "regexp"
  19         "strings"
  20         "sync"
  21         "syscall"
  22         "time"
  23
  24         "git.curoverse.com/arvados.git/sdk/go/arvados"
  25         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  26         "git.curoverse.com/arvados.git/sdk/go/config"
  27         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  28         "github.com/coreos/go-systemd/daemon"
  29         "github.com/ghodss/yaml"
  30         "github.com/gorilla/mux"
  31 )
  32
  33 type Config struct {
  34         Client          arvados.Client
  35         Listen          string
  36         DisableGet      bool
  37         DisablePut      bool
  38         DefaultReplicas int
  39         Timeout         arvados.Duration
  40         PIDFile         string
  41         Debug           bool
  42 }
  43
  44 func DefaultConfig() *Config {
  45         return &Config{
  46                 Listen:  ":25107",
  47                 Timeout: arvados.Duration(15 * time.Second),
  48         }
  49 }
  50
  51 var (
  52         listener net.Listener
  53         router   http.Handler
  54 )
  55
  56 func main() {
  57         cfg := DefaultConfig()
  58
  59         flagset := flag.NewFlagSet("keepproxy", flag.ExitOnError)
  60         flagset.Usage = usage
  61
  62         const deprecated = " (DEPRECATED -- use config file instead)"
  63         flagset.StringVar(&cfg.Listen, "listen", cfg.Listen, "Local port to listen on."+deprecated)
  64         flagset.BoolVar(&cfg.DisableGet, "no-get", cfg.DisableGet, "Disable GET operations."+deprecated)
  65         flagset.BoolVar(&cfg.DisablePut, "no-put", cfg.DisablePut, "Disable PUT operations."+deprecated)
  66         flagset.IntVar(&cfg.DefaultReplicas, "default-replicas", cfg.DefaultReplicas, "Default number of replicas to write if not specified by the client. If 0, use site default."+deprecated)
  67         flagset.StringVar(&cfg.PIDFile, "pid", cfg.PIDFile, "Path to write pid file."+deprecated)
  68         timeoutSeconds := flagset.Int("timeout", int(time.Duration(cfg.Timeout)/time.Second), "Timeout (in seconds) on requests to internal Keep services."+deprecated)
  69
  70         var cfgPath string
  71         const defaultCfgPath = "/etc/arvados/keepproxy/keepproxy.yml"
  72         flagset.StringVar(&cfgPath, "config", defaultCfgPath, "Configuration file `path`")
  73         dumpConfig := flagset.Bool("dump-config", false, "write current configuration to stdout and exit")
  74         flagset.Parse(os.Args[1:])
  75
  76         err := config.LoadFile(cfg, cfgPath)
  77         if err != nil {
  78                 h := os.Getenv("ARVADOS_API_HOST")
  79                 t := os.Getenv("ARVADOS_API_TOKEN")
  80                 if h == "" || t == "" || !os.IsNotExist(err) || cfgPath != defaultCfgPath {
  81                         log.Fatal(err)
  82                 }
  83                 log.Print("DEPRECATED: No config file found, but ARVADOS_API_HOST and ARVADOS_API_TOKEN environment variables are set. Please use a config file instead.")
  84                 cfg.Client.APIHost = h
  85                 cfg.Client.AuthToken = t
  86                 if regexp.MustCompile("^(?i:1|yes|true)$").MatchString(os.Getenv("ARVADOS_API_HOST_INSECURE")) {
  87                         cfg.Client.Insecure = true
  88                 }
  89                 if y, err := yaml.Marshal(cfg); err == nil && !*dumpConfig {
  90                         log.Print("Current configuration:\n", string(y))
  91                 }
  92                 cfg.Timeout = arvados.Duration(time.Duration(*timeoutSeconds) * time.Second)
  93         }
  94
  95         if *dumpConfig {
  96                 log.Fatal(config.DumpAndExit(cfg))
  97         }
  98
  99         arv, err := arvadosclient.New(&cfg.Client)
 100         if err != nil {
 101                 log.Fatalf("Error setting up arvados client %s", err.Error())
 102         }
 103
 104         if cfg.Debug {
 105                 keepclient.DebugPrintf = log.Printf
 106         }
 107         kc, err := keepclient.MakeKeepClient(arv)
 108         if err != nil {
 109                 log.Fatalf("Error setting up keep client %s", err.Error())
 110         }
 111         keepclient.RefreshServiceDiscoveryOnSIGHUP()
 112
 113         if cfg.PIDFile != "" {
 114                 f, err := os.Create(cfg.PIDFile)
 115                 if err != nil {
 116                         log.Fatal(err)
 117                 }
 118                 defer f.Close()
 119                 err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
 120                 if err != nil {
 121                         log.Fatalf("flock(%s): %s", cfg.PIDFile, err)
 122                 }
 123                 defer os.Remove(cfg.PIDFile)
 124                 err = f.Truncate(0)
 125                 if err != nil {
 126                         log.Fatalf("truncate(%s): %s", cfg.PIDFile, err)
 127                 }
 128                 _, err = fmt.Fprint(f, os.Getpid())
 129                 if err != nil {
 130                         log.Fatalf("write(%s): %s", cfg.PIDFile, err)
 131                 }
 132                 err = f.Sync()
 133                 if err != nil {
 134                         log.Fatal("sync(%s): %s", cfg.PIDFile, err)
 135                 }
 136         }
 137
 138         if cfg.DefaultReplicas > 0 {
 139                 kc.Want_replicas = cfg.DefaultReplicas
 140         }
 141
 142         listener, err = net.Listen("tcp", cfg.Listen)
 143         if err != nil {
 144                 log.Fatalf("listen(%s): %s", cfg.Listen, err)
 145         }
 146         if _, err := daemon.SdNotify(false, "READY=1"); err != nil {
 147                 log.Printf("Error notifying init daemon: %v", err)
 148         }
 149         log.Println("Listening at", listener.Addr())
 150
 151         // Shut down the server gracefully (by closing the listener)
 152         // if SIGTERM is received.
 153         term := make(chan os.Signal, 1)
 154         go func(sig <-chan os.Signal) {
 155                 s := <-sig
 156                 log.Println("caught signal:", s)
 157                 listener.Close()
 158         }(term)
 159         signal.Notify(term, syscall.SIGTERM)
 160         signal.Notify(term, syscall.SIGINT)
 161
 162         // Start serving requests.
 163         router = MakeRESTRouter(!cfg.DisableGet, !cfg.DisablePut, kc, time.Duration(cfg.Timeout))
 164         http.Serve(listener, router)
 165
 166         log.Println("shutting down")
 167 }
 168
 169 type ApiTokenCache struct {
 170         tokens     map[string]int64
 171         lock       sync.Mutex
 172         expireTime int64
 173 }
 174
 175 // Cache the token and set an expire time.  If we already have an expire time
 176 // on the token, it is not updated.
 177 func (this *ApiTokenCache) RememberToken(token string) {
 178         this.lock.Lock()
 179         defer this.lock.Unlock()
 180
 181         now := time.Now().Unix()
 182         if this.tokens[token] == 0 {
 183                 this.tokens[token] = now + this.expireTime
 184         }
 185 }
 186
 187 // Check if the cached token is known and still believed to be valid.
 188 func (this *ApiTokenCache) RecallToken(token string) bool {
 189         this.lock.Lock()
 190         defer this.lock.Unlock()
 191
 192         now := time.Now().Unix()
 193         if this.tokens[token] == 0 {
 194                 // Unknown token
 195                 return false
 196         } else if now < this.tokens[token] {
 197                 // Token is known and still valid
 198                 return true
 199         } else {
 200                 // Token is expired
 201                 this.tokens[token] = 0
 202                 return false
 203         }
 204 }
 205
 206 func GetRemoteAddress(req *http.Request) string {
 207         if xff := req.Header.Get("X-Forwarded-For"); xff != "" {
 208                 return xff + "," + req.RemoteAddr
 209         }
 210         return req.RemoteAddr
 211 }
 212
 213 func CheckAuthorizationHeader(kc *keepclient.KeepClient, cache *ApiTokenCache, req *http.Request) (pass bool, tok string) {
 214         var auth string
 215         if auth = req.Header.Get("Authorization"); auth == "" {
 216                 return false, ""
 217         }
 218
 219         _, err := fmt.Sscanf(auth, "OAuth2 %s", &tok)
 220         if err != nil {
 221                 // Scanning error
 222                 return false, ""
 223         }
 224
 225         if cache.RecallToken(tok) {
 226                 // Valid in the cache, short circuit
 227                 return true, tok
 228         }
 229
 230         arv := *kc.Arvados
 231         arv.ApiToken = tok
 232         if err := arv.Call("HEAD", "users", "", "current", nil, nil); err != nil {
 233                 log.Printf("%s: CheckAuthorizationHeader error: %v", GetRemoteAddress(req), err)
 234                 return false, ""
 235         }
 236
 237         // Success!  Update cache
 238         cache.RememberToken(tok)
 239
 240         return true, tok
 241 }
 242
 243 type proxyHandler struct {
 244         http.Handler
 245         *keepclient.KeepClient
 246         *ApiTokenCache
 247         timeout   time.Duration
 248         transport *http.Transport
 249 }
 250
 251 // MakeRESTRouter returns an http.Handler that passes GET and PUT
 252 // requests to the appropriate handlers.
 253 func MakeRESTRouter(enable_get bool, enable_put bool, kc *keepclient.KeepClient, timeout time.Duration) http.Handler {
 254         rest := mux.NewRouter()
 255
 256         transport := *(http.DefaultTransport.(*http.Transport))
 257         transport.DialContext = (&net.Dialer{
 258                 Timeout:   keepclient.DefaultConnectTimeout,
 259                 KeepAlive: keepclient.DefaultKeepAlive,
 260                 DualStack: true,
 261         }).DialContext
 262         transport.TLSClientConfig = arvadosclient.MakeTLSConfig(kc.Arvados.ApiInsecure)
 263         transport.TLSHandshakeTimeout = keepclient.DefaultTLSHandshakeTimeout
 264
 265         h := &proxyHandler{
 266                 Handler:    rest,
 267                 KeepClient: kc,
 268                 timeout:    timeout,
 269                 transport:  &transport,
 270                 ApiTokenCache: &ApiTokenCache{
 271                         tokens:     make(map[string]int64),
 272                         expireTime: 300,
 273                 },
 274         }
 275
 276         if enable_get {
 277                 rest.HandleFunc(`/{locator:[0-9a-f]{32}\+.*}`, h.Get).Methods("GET", "HEAD")
 278                 rest.HandleFunc(`/{locator:[0-9a-f]{32}}`, h.Get).Methods("GET", "HEAD")
 279
 280                 // List all blocks
 281                 rest.HandleFunc(`/index`, h.Index).Methods("GET")
 282
 283                 // List blocks whose hash has the given prefix
 284                 rest.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, h.Index).Methods("GET")
 285         }
 286
 287         if enable_put {
 288                 rest.HandleFunc(`/{locator:[0-9a-f]{32}\+.*}`, h.Put).Methods("PUT")
 289                 rest.HandleFunc(`/{locator:[0-9a-f]{32}}`, h.Put).Methods("PUT")
 290                 rest.HandleFunc(`/`, h.Put).Methods("POST")
 291                 rest.HandleFunc(`/{any}`, h.Options).Methods("OPTIONS")
 292                 rest.HandleFunc(`/`, h.Options).Methods("OPTIONS")
 293         }
 294
 295         rest.NotFoundHandler = InvalidPathHandler{}
 296         return h
 297 }
 298
 299 var errLoopDetected = errors.New("loop detected")
 300
 301 func (*proxyHandler) checkLoop(resp http.ResponseWriter, req *http.Request) error {
 302         if via := req.Header.Get("Via"); strings.Index(via, " "+viaAlias) >= 0 {
 303                 log.Printf("proxy loop detected (request has Via: %q): perhaps keepproxy is misidentified by gateway config as an external client, or its keep_services record does not have service_type=proxy?", via)
 304                 http.Error(resp, errLoopDetected.Error(), http.StatusInternalServerError)
 305                 return errLoopDetected
 306         }
 307         return nil
 308 }
 309
 310 func SetCorsHeaders(resp http.ResponseWriter) {
 311         resp.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, POST, PUT, OPTIONS")
 312         resp.Header().Set("Access-Control-Allow-Origin", "*")
 313         resp.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Length, Content-Type, X-Keep-Desired-Replicas")
 314         resp.Header().Set("Access-Control-Max-Age", "86486400")
 315 }
 316
 317 type InvalidPathHandler struct{}
 318
 319 func (InvalidPathHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 320         log.Printf("%s: %s %s unroutable", GetRemoteAddress(req), req.Method, req.URL.Path)
 321         http.Error(resp, "Bad request", http.StatusBadRequest)
 322 }
 323
 324 func (h *proxyHandler) Options(resp http.ResponseWriter, req *http.Request) {
 325         log.Printf("%s: %s %s", GetRemoteAddress(req), req.Method, req.URL.Path)
 326         SetCorsHeaders(resp)
 327 }
 328
 329 var BadAuthorizationHeader = errors.New("Missing or invalid Authorization header")
 330 var ContentLengthMismatch = errors.New("Actual length != expected content length")
 331 var MethodNotSupported = errors.New("Method not supported")
 332
 333 var removeHint, _ = regexp.Compile("\\+K@[a-z0-9]{5}(\\+|$)")
 334
 335 func (h *proxyHandler) Get(resp http.ResponseWriter, req *http.Request) {
 336         if err := h.checkLoop(resp, req); err != nil {
 337                 return
 338         }
 339         SetCorsHeaders(resp)
 340         resp.Header().Set("Via", req.Proto+" "+viaAlias)
 341
 342         locator := mux.Vars(req)["locator"]
 343         var err error
 344         var status int
 345         var expectLength, responseLength int64
 346         var proxiedURI = "-"
 347
 348         defer func() {
 349                 log.Println(GetRemoteAddress(req), req.Method, req.URL.Path, status, expectLength, responseLength, proxiedURI, err)
 350                 if status != http.StatusOK {
 351                         http.Error(resp, err.Error(), status)
 352                 }
 353         }()
 354
 355         kc := h.makeKeepClient(req)
 356
 357         var pass bool
 358         var tok string
 359         if pass, tok = CheckAuthorizationHeader(kc, h.ApiTokenCache, req); !pass {
 360                 status, err = http.StatusForbidden, BadAuthorizationHeader
 361                 return
 362         }
 363
 364         // Copy ArvadosClient struct and use the client's API token
 365         arvclient := *kc.Arvados
 366         arvclient.ApiToken = tok
 367         kc.Arvados = &arvclient
 368
 369         var reader io.ReadCloser
 370
 371         locator = removeHint.ReplaceAllString(locator, "$1")
 372
 373         switch req.Method {
 374         case "HEAD":
 375                 expectLength, proxiedURI, err = kc.Ask(locator)
 376         case "GET":
 377                 reader, expectLength, proxiedURI, err = kc.Get(locator)
 378                 if reader != nil {
 379                         defer reader.Close()
 380                 }
 381         default:
 382                 status, err = http.StatusNotImplemented, MethodNotSupported
 383                 return
 384         }
 385
 386         if expectLength == -1 {
 387                 log.Println("Warning:", GetRemoteAddress(req), req.Method, proxiedURI, "Content-Length not provided")
 388         }
 389
 390         switch respErr := err.(type) {
 391         case nil:
 392                 status = http.StatusOK
 393                 resp.Header().Set("Content-Length", fmt.Sprint(expectLength))
 394                 switch req.Method {
 395                 case "HEAD":
 396                         responseLength = 0
 397                 case "GET":
 398                         responseLength, err = io.Copy(resp, reader)
 399                         if err == nil && expectLength > -1 && responseLength != expectLength {
 400                                 err = ContentLengthMismatch
 401                         }
 402                 }
 403         case keepclient.Error:
 404                 if respErr == keepclient.BlockNotFound {
 405                         status = http.StatusNotFound
 406                 } else if respErr.Temporary() {
 407                         status = http.StatusBadGateway
 408                 } else {
 409                         status = 422
 410                 }
 411         default:
 412                 status = http.StatusInternalServerError
 413         }
 414 }
 415
 416 var LengthRequiredError = errors.New(http.StatusText(http.StatusLengthRequired))
 417 var LengthMismatchError = errors.New("Locator size hint does not match Content-Length header")
 418
 419 func (h *proxyHandler) Put(resp http.ResponseWriter, req *http.Request) {
 420         if err := h.checkLoop(resp, req); err != nil {
 421                 return
 422         }
 423         SetCorsHeaders(resp)
 424         resp.Header().Set("Via", "HTTP/1.1 "+viaAlias)
 425
 426         kc := h.makeKeepClient(req)
 427
 428         var err error
 429         var expectLength int64
 430         var status = http.StatusInternalServerError
 431         var wroteReplicas int
 432         var locatorOut string = "-"
 433
 434         defer func() {
 435                 log.Println(GetRemoteAddress(req), req.Method, req.URL.Path, status, expectLength, kc.Want_replicas, wroteReplicas, locatorOut, err)
 436                 if status != http.StatusOK {
 437                         http.Error(resp, err.Error(), status)
 438                 }
 439         }()
 440
 441         locatorIn := mux.Vars(req)["locator"]
 442
 443         _, err = fmt.Sscanf(req.Header.Get("Content-Length"), "%d", &expectLength)
 444         if err != nil || expectLength < 0 {
 445                 err = LengthRequiredError
 446                 status = http.StatusLengthRequired
 447                 return
 448         }
 449
 450         if locatorIn != "" {
 451                 var loc *keepclient.Locator
 452                 if loc, err = keepclient.MakeLocator(locatorIn); err != nil {
 453                         status = http.StatusBadRequest
 454                         return
 455                 } else if loc.Size > 0 && int64(loc.Size) != expectLength {
 456                         err = LengthMismatchError
 457                         status = http.StatusBadRequest
 458                         return
 459                 }
 460         }
 461
 462         var pass bool
 463         var tok string
 464         if pass, tok = CheckAuthorizationHeader(kc, h.ApiTokenCache, req); !pass {
 465                 err = BadAuthorizationHeader
 466                 status = http.StatusForbidden
 467                 return
 468         }
 469
 470         // Copy ArvadosClient struct and use the client's API token
 471         arvclient := *kc.Arvados
 472         arvclient.ApiToken = tok
 473         kc.Arvados = &arvclient
 474
 475         // Check if the client specified the number of replicas
 476         if req.Header.Get("X-Keep-Desired-Replicas") != "" {
 477                 var r int
 478                 _, err := fmt.Sscanf(req.Header.Get(keepclient.X_Keep_Desired_Replicas), "%d", &r)
 479                 if err == nil {
 480                         kc.Want_replicas = r
 481                 }
 482         }
 483
 484         // Now try to put the block through
 485         if locatorIn == "" {
 486                 if bytes, err := ioutil.ReadAll(req.Body); err != nil {
 487                         err = errors.New(fmt.Sprintf("Error reading request body: %s", err))
 488                         status = http.StatusInternalServerError
 489                         return
 490                 } else {
 491                         locatorOut, wroteReplicas, err = kc.PutB(bytes)
 492                 }
 493         } else {
 494                 locatorOut, wroteReplicas, err = kc.PutHR(locatorIn, req.Body, expectLength)
 495         }
 496
 497         // Tell the client how many successful PUTs we accomplished
 498         resp.Header().Set(keepclient.X_Keep_Replicas_Stored, fmt.Sprintf("%d", wroteReplicas))
 499
 500         switch err.(type) {
 501         case nil:
 502                 status = http.StatusOK
 503                 _, err = io.WriteString(resp, locatorOut)
 504
 505         case keepclient.OversizeBlockError:
 506                 // Too much data
 507                 status = http.StatusRequestEntityTooLarge
 508
 509         case keepclient.InsufficientReplicasError:
 510                 if wroteReplicas > 0 {
 511                         // At least one write is considered success.  The
 512                         // client can decide if getting less than the number of
 513                         // replications it asked for is a fatal error.
 514                         status = http.StatusOK
 515                         _, err = io.WriteString(resp, locatorOut)
 516                 } else {
 517                         status = http.StatusServiceUnavailable
 518                 }
 519
 520         default:
 521                 status = http.StatusBadGateway
 522         }
 523 }
 524
 525 // ServeHTTP implementation for IndexHandler
 526 // Supports only GET requests for /index/{prefix:[0-9a-f]{0,32}}
 527 // For each keep server found in LocalRoots:
 528 //   Invokes GetIndex using keepclient
 529 //   Expects "complete" response (terminating with blank new line)
 530 //   Aborts on any errors
 531 // Concatenates responses from all those keep servers and returns
 532 func (h *proxyHandler) Index(resp http.ResponseWriter, req *http.Request) {
 533         SetCorsHeaders(resp)
 534
 535         prefix := mux.Vars(req)["prefix"]
 536         var err error
 537         var status int
 538
 539         defer func() {
 540                 if status != http.StatusOK {
 541                         http.Error(resp, err.Error(), status)
 542                 }
 543         }()
 544
 545         kc := h.makeKeepClient(req)
 546         ok, token := CheckAuthorizationHeader(kc, h.ApiTokenCache, req)
 547         if !ok {
 548                 status, err = http.StatusForbidden, BadAuthorizationHeader
 549                 return
 550         }
 551
 552         // Copy ArvadosClient struct and use the client's API token
 553         arvclient := *kc.Arvados
 554         arvclient.ApiToken = token
 555         kc.Arvados = &arvclient
 556
 557         // Only GET method is supported
 558         if req.Method != "GET" {
 559                 status, err = http.StatusNotImplemented, MethodNotSupported
 560                 return
 561         }
 562
 563         // Get index from all LocalRoots and write to resp
 564         var reader io.Reader
 565         for uuid := range kc.LocalRoots() {
 566                 reader, err = kc.GetIndex(uuid, prefix)
 567                 if err != nil {
 568                         status = http.StatusBadGateway
 569                         return
 570                 }
 571
 572                 _, err = io.Copy(resp, reader)
 573                 if err != nil {
 574                         status = http.StatusBadGateway
 575                         return
 576                 }
 577         }
 578
 579         // Got index from all the keep servers and wrote to resp
 580         status = http.StatusOK
 581         resp.Write([]byte("\n"))
 582 }
 583
 584 func (h *proxyHandler) makeKeepClient(req *http.Request) *keepclient.KeepClient {
 585         kc := *h.KeepClient
 586         kc.HTTPClient = &proxyClient{
 587                 client: &http.Client{
 588                         Timeout:   h.timeout,
 589                         Transport: h.transport,
 590                 },
 591                 proto: req.Proto,
 592         }
 593         return &kc
 594 }