services/keep-web/s3.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         "encoding/xml"
   9         "errors"
  10         "fmt"
  11         "io"
  12         "net/http"
  13         "os"
  14         "path/filepath"
  15         "sort"
  16         "strconv"
  17         "strings"
  18
  19         "git.arvados.org/arvados.git/sdk/go/arvados"
  20         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  21         "github.com/AdRoll/goamz/s3"
  22 )
  23
  24 const s3MaxKeys = 1000
  25
  26 // serveS3 handles r and returns true if r is a request from an S3
  27 // client, otherwise it returns false.
  28 func (h *handler) serveS3(w http.ResponseWriter, r *http.Request) bool {
  29         var token string
  30         if auth := r.Header.Get("Authorization"); strings.HasPrefix(auth, "AWS ") {
  31                 split := strings.SplitN(auth[4:], ":", 2)
  32                 if len(split) < 2 {
  33                         w.WriteHeader(http.StatusUnauthorized)
  34                         return true
  35                 }
  36                 token = split[0]
  37         } else if strings.HasPrefix(auth, "AWS4-HMAC-SHA256 ") {
  38                 for _, cmpt := range strings.Split(auth[17:], ",") {
  39                         cmpt = strings.TrimSpace(cmpt)
  40                         split := strings.SplitN(cmpt, "=", 2)
  41                         if len(split) == 2 && split[0] == "Credential" {
  42                                 keyandscope := strings.Split(split[1], "/")
  43                                 if len(keyandscope[0]) > 0 {
  44                                         token = keyandscope[0]
  45                                         break
  46                                 }
  47                         }
  48                 }
  49                 if token == "" {
  50                         w.WriteHeader(http.StatusBadRequest)
  51                         fmt.Println(w, "invalid V4 signature")
  52                         return true
  53                 }
  54         } else {
  55                 return false
  56         }
  57
  58         _, kc, client, release, err := h.getClients(r.Header.Get("X-Request-Id"), token)
  59         if err != nil {
  60                 http.Error(w, "Pool failed: "+h.clientPool.Err().Error(), http.StatusInternalServerError)
  61                 return true
  62         }
  63         defer release()
  64
  65         fs := client.SiteFileSystem(kc)
  66         fs.ForwardSlashNameSubstitution(h.Config.cluster.Collections.ForwardSlashNameSubstitution)
  67
  68         objectNameGiven := strings.Count(strings.TrimSuffix(r.URL.Path, "/"), "/") > 1
  69
  70         switch {
  71         case r.Method == "GET" && !objectNameGiven:
  72                 // Path is "/{uuid}" or "/{uuid}/", has no object name
  73                 if _, ok := r.URL.Query()["versioning"]; ok {
  74                         // GetBucketVersioning
  75                         w.Header().Set("Content-Type", "application/xml")
  76                         io.WriteString(w, xml.Header)
  77                         fmt.Fprintln(w, `<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"/>`)
  78                 } else {
  79                         // ListObjects
  80                         h.s3list(w, r, fs)
  81                 }
  82                 return true
  83         case r.Method == "GET" || r.Method == "HEAD":
  84                 fspath := "/by_id" + r.URL.Path
  85                 fi, err := fs.Stat(fspath)
  86                 if r.Method == "HEAD" && !objectNameGiven {
  87                         // HeadBucket
  88                         if err == nil && fi.IsDir() {
  89                                 w.WriteHeader(http.StatusOK)
  90                         } else if os.IsNotExist(err) {
  91                                 w.WriteHeader(http.StatusNotFound)
  92                         } else {
  93                                 http.Error(w, err.Error(), http.StatusBadGateway)
  94                         }
  95                         return true
  96                 }
  97                 if err == nil && fi.IsDir() && objectNameGiven && strings.HasSuffix(fspath, "/") && h.Config.cluster.Collections.S3FolderObjects {
  98                         w.Header().Set("Content-Type", "application/x-directory")
  99                         w.WriteHeader(http.StatusOK)
 100                         return true
 101                 }
 102                 if os.IsNotExist(err) ||
 103                         (err != nil && err.Error() == "not a directory") ||
 104                         (fi != nil && fi.IsDir()) {
 105                         http.Error(w, "not found", http.StatusNotFound)
 106                         return true
 107                 }
 108                 // shallow copy r, and change URL path
 109                 r := *r
 110                 r.URL.Path = fspath
 111                 http.FileServer(fs).ServeHTTP(w, &r)
 112                 return true
 113         case r.Method == "PUT":
 114                 if !objectNameGiven {
 115                         http.Error(w, "missing object name in PUT request", http.StatusBadRequest)
 116                         return true
 117                 }
 118                 fspath := "by_id" + r.URL.Path
 119                 var objectIsDir bool
 120                 if strings.HasSuffix(fspath, "/") {
 121                         if !h.Config.cluster.Collections.S3FolderObjects {
 122                                 http.Error(w, "invalid object name: trailing slash", http.StatusBadRequest)
 123                                 return true
 124                         }
 125                         n, err := r.Body.Read(make([]byte, 1))
 126                         if err != nil && err != io.EOF {
 127                                 http.Error(w, fmt.Sprintf("error reading request body: %s", err), http.StatusInternalServerError)
 128                                 return true
 129                         } else if n > 0 {
 130                                 http.Error(w, "cannot write a non-empty file with a trailing '/' char", http.StatusBadRequest)
 131                                 return true
 132                         }
 133                         // Given PUT "foo/bar/", we'll use "foo/bar/."
 134                         // in the "ensure parents exist" block below,
 135                         // and then we'll be done.
 136                         fspath += "."
 137                         objectIsDir = true
 138                 }
 139                 fi, err := fs.Stat(fspath)
 140                 if err != nil && err.Error() == "not a directory" {
 141                         // requested foo/bar, but foo is a file
 142                         http.Error(w, "object name conflicts with existing object", http.StatusBadRequest)
 143                         return true
 144                 }
 145                 if strings.HasSuffix(r.URL.Path, "/") && err == nil && !fi.IsDir() {
 146                         // requested foo/bar/, but foo/bar is a file
 147                         http.Error(w, "object name conflicts with existing object", http.StatusBadRequest)
 148                         return true
 149                 }
 150                 // create missing parent/intermediate directories, if any
 151                 for i, c := range fspath {
 152                         if i > 0 && c == '/' {
 153                                 dir := fspath[:i]
 154                                 if strings.HasSuffix(dir, "/") {
 155                                         err = errors.New("invalid object name (consecutive '/' chars)")
 156                                         http.Error(w, err.Error(), http.StatusBadRequest)
 157                                         return true
 158                                 }
 159                                 err := fs.Mkdir(dir, 0755)
 160                                 if err != nil && err != os.ErrExist {
 161                                         err = fmt.Errorf("mkdir %q failed: %w", dir, err)
 162                                         http.Error(w, err.Error(), http.StatusInternalServerError)
 163                                         return true
 164                                 }
 165                         }
 166                 }
 167                 if !objectIsDir {
 168                         f, err := fs.OpenFile(fspath, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
 169                         if os.IsNotExist(err) {
 170                                 f, err = fs.OpenFile(fspath, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
 171                         }
 172                         if err != nil {
 173                                 err = fmt.Errorf("open %q failed: %w", r.URL.Path, err)
 174                                 http.Error(w, err.Error(), http.StatusBadRequest)
 175                                 return true
 176                         }
 177                         defer f.Close()
 178                         _, err = io.Copy(f, r.Body)
 179                         if err != nil {
 180                                 err = fmt.Errorf("write to %q failed: %w", r.URL.Path, err)
 181                                 http.Error(w, err.Error(), http.StatusBadGateway)
 182                                 return true
 183                         }
 184                         err = f.Close()
 185                         if err != nil {
 186                                 err = fmt.Errorf("write to %q failed: close: %w", r.URL.Path, err)
 187                                 http.Error(w, err.Error(), http.StatusBadGateway)
 188                                 return true
 189                         }
 190                 }
 191                 err = fs.Sync()
 192                 if err != nil {
 193                         err = fmt.Errorf("sync failed: %w", err)
 194                         http.Error(w, err.Error(), http.StatusInternalServerError)
 195                         return true
 196                 }
 197                 w.WriteHeader(http.StatusOK)
 198                 return true
 199         default:
 200                 http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
 201                 return true
 202         }
 203 }
 204
 205 // Call fn on the given path (directory) and its contents, in
 206 // lexicographic order.
 207 //
 208 // If isRoot==true and path is not a directory, return nil.
 209 //
 210 // If fn returns filepath.SkipDir when called on a directory, don't
 211 // descend into that directory.
 212 func walkFS(fs arvados.CustomFileSystem, path string, isRoot bool, fn func(path string, fi os.FileInfo) error) error {
 213         if isRoot {
 214                 fi, err := fs.Stat(path)
 215                 if os.IsNotExist(err) || (err == nil && !fi.IsDir()) {
 216                         return nil
 217                 } else if err != nil {
 218                         return err
 219                 }
 220                 err = fn(path, fi)
 221                 if err == filepath.SkipDir {
 222                         return nil
 223                 } else if err != nil {
 224                         return err
 225                 }
 226         }
 227         f, err := fs.Open(path)
 228         if os.IsNotExist(err) && isRoot {
 229                 return nil
 230         } else if err != nil {
 231                 return fmt.Errorf("open %q: %w", path, err)
 232         }
 233         defer f.Close()
 234         if path == "/" {
 235                 path = ""
 236         }
 237         fis, err := f.Readdir(-1)
 238         if err != nil {
 239                 return err
 240         }
 241         sort.Slice(fis, func(i, j int) bool { return fis[i].Name() < fis[j].Name() })
 242         for _, fi := range fis {
 243                 err = fn(path+"/"+fi.Name(), fi)
 244                 if err == filepath.SkipDir {
 245                         continue
 246                 } else if err != nil {
 247                         return err
 248                 }
 249                 if fi.IsDir() {
 250                         err = walkFS(fs, path+"/"+fi.Name(), false, fn)
 251                         if err != nil {
 252                                 return err
 253                         }
 254                 }
 255         }
 256         return nil
 257 }
 258
 259 var errDone = errors.New("done")
 260
 261 func (h *handler) s3list(w http.ResponseWriter, r *http.Request, fs arvados.CustomFileSystem) {
 262         var params struct {
 263                 bucket    string
 264                 delimiter string
 265                 marker    string
 266                 maxKeys   int
 267                 prefix    string
 268         }
 269         params.bucket = strings.SplitN(r.URL.Path[1:], "/", 2)[0]
 270         params.delimiter = r.FormValue("delimiter")
 271         params.marker = r.FormValue("marker")
 272         if mk, _ := strconv.ParseInt(r.FormValue("max-keys"), 10, 64); mk > 0 && mk < s3MaxKeys {
 273                 params.maxKeys = int(mk)
 274         } else {
 275                 params.maxKeys = s3MaxKeys
 276         }
 277         params.prefix = r.FormValue("prefix")
 278
 279         bucketdir := "by_id/" + params.bucket
 280         // walkpath is the directory (relative to bucketdir) we need
 281         // to walk: the innermost directory that is guaranteed to
 282         // contain all paths that have the requested prefix. Examples:
 283         // prefix "foo/bar"  => walkpath "foo"
 284         // prefix "foo/bar/" => walkpath "foo/bar"
 285         // prefix "foo"      => walkpath ""
 286         // prefix ""         => walkpath ""
 287         walkpath := params.prefix
 288         if cut := strings.LastIndex(walkpath, "/"); cut >= 0 {
 289                 walkpath = walkpath[:cut]
 290         } else {
 291                 walkpath = ""
 292         }
 293
 294         resp := s3.ListResp{
 295                 Name:      strings.SplitN(r.URL.Path[1:], "/", 2)[0],
 296                 Prefix:    params.prefix,
 297                 Delimiter: params.delimiter,
 298                 Marker:    params.marker,
 299                 MaxKeys:   params.maxKeys,
 300         }
 301         commonPrefixes := map[string]bool{}
 302         err := walkFS(fs, strings.TrimSuffix(bucketdir+"/"+walkpath, "/"), true, func(path string, fi os.FileInfo) error {
 303                 if path == bucketdir {
 304                         return nil
 305                 }
 306                 path = path[len(bucketdir)+1:]
 307                 filesize := fi.Size()
 308                 if fi.IsDir() {
 309                         path += "/"
 310                         filesize = 0
 311                 }
 312                 if len(path) <= len(params.prefix) {
 313                         if path > params.prefix[:len(path)] {
 314                                 // with prefix "foobar", walking "fooz" means we're done
 315                                 return errDone
 316                         }
 317                         if path < params.prefix[:len(path)] {
 318                                 // with prefix "foobar", walking "foobag" is pointless
 319                                 return filepath.SkipDir
 320                         }
 321                         if fi.IsDir() && !strings.HasPrefix(params.prefix+"/", path) {
 322                                 // with prefix "foo/bar", walking "fo"
 323                                 // is pointless (but walking "foo" or
 324                                 // "foo/bar" is necessary)
 325                                 return filepath.SkipDir
 326                         }
 327                         if len(path) < len(params.prefix) {
 328                                 // can't skip anything, and this entry
 329                                 // isn't in the results, so just
 330                                 // continue descent
 331                                 return nil
 332                         }
 333                 } else {
 334                         if path[:len(params.prefix)] > params.prefix {
 335                                 // with prefix "foobar", nothing we
 336                                 // see after "foozzz" is relevant
 337                                 return errDone
 338                         }
 339                 }
 340                 if path < params.marker || path < params.prefix {
 341                         return nil
 342                 }
 343                 if fi.IsDir() && !h.Config.cluster.Collections.S3FolderObjects {
 344                         // Note we don't add anything to
 345                         // commonPrefixes here even if delimiter is
 346                         // "/". We descend into the directory, and
 347                         // return a commonPrefix only if we end up
 348                         // finding a regular file inside it.
 349                         return nil
 350                 }
 351                 if params.delimiter != "" {
 352                         idx := strings.Index(path[len(params.prefix):], params.delimiter)
 353                         if idx >= 0 {
 354                                 // with prefix "foobar" and delimiter
 355                                 // "z", when we hit "foobar/baz", we
 356                                 // add "/baz" to commonPrefixes and
 357                                 // stop descending.
 358                                 commonPrefixes[path[:len(params.prefix)+idx+1]] = true
 359                                 return filepath.SkipDir
 360                         }
 361                 }
 362                 if len(resp.Contents)+len(commonPrefixes) >= params.maxKeys {
 363                         resp.IsTruncated = true
 364                         if params.delimiter != "" {
 365                                 resp.NextMarker = path
 366                         }
 367                         return errDone
 368                 }
 369                 resp.Contents = append(resp.Contents, s3.Key{
 370                         Key:          path,
 371                         LastModified: fi.ModTime().UTC().Format("2006-01-02T15:04:05.999") + "Z",
 372                         Size:         filesize,
 373                 })
 374                 return nil
 375         })
 376         if err != nil && err != errDone {
 377                 http.Error(w, err.Error(), http.StatusInternalServerError)
 378                 return
 379         }
 380         if params.delimiter != "" {
 381                 for prefix := range commonPrefixes {
 382                         resp.CommonPrefixes = append(resp.CommonPrefixes, prefix)
 383                         sort.Strings(resp.CommonPrefixes)
 384                 }
 385         }
 386         wrappedResp := struct {
 387                 XMLName string `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListBucketResult"`
 388                 s3.ListResp
 389         }{"", resp}
 390         w.Header().Set("Content-Type", "application/xml")
 391         io.WriteString(w, xml.Header)
 392         if err := xml.NewEncoder(w).Encode(wrappedResp); err != nil {
 393                 ctxlog.FromContext(r.Context()).WithError(err).Error("error writing xml response")
 394         }
 395 }