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         switch {
  69         case r.Method == "GET" && strings.Count(strings.TrimSuffix(r.URL.Path, "/"), "/") == 1:
  70                 // Path is "/{uuid}" or "/{uuid}/", has no object name
  71                 if _, ok := r.URL.Query()["versioning"]; ok {
  72                         // GetBucketVersioning
  73                         w.Header().Set("Content-Type", "application/xml")
  74                         io.WriteString(w, xml.Header)
  75                         fmt.Fprintln(w, `<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"/>`)
  76                 } else {
  77                         // ListObjects
  78                         h.s3list(w, r, fs)
  79                 }
  80                 return true
  81         case r.Method == "GET":
  82                 fspath := "/by_id" + r.URL.Path
  83                 fi, err := fs.Stat(fspath)
  84                 if os.IsNotExist(err) ||
  85                         (err != nil && err.Error() == "not a directory") ||
  86                         (fi != nil && fi.IsDir()) {
  87                         http.Error(w, "not found", http.StatusNotFound)
  88                         return true
  89                 }
  90                 // shallow copy r, and change URL path
  91                 r := *r
  92                 r.URL.Path = fspath
  93                 http.FileServer(fs).ServeHTTP(w, &r)
  94                 return true
  95         case r.Method == "PUT":
  96                 if strings.HasSuffix(r.URL.Path, "/") {
  97                         http.Error(w, "invalid object name (trailing '/' char)", http.StatusBadRequest)
  98                         return true
  99                 }
 100                 fspath := "by_id" + r.URL.Path
 101                 _, err = fs.Stat(fspath)
 102                 if err != nil && err.Error() == "not a directory" {
 103                         // requested foo/bar, but foo is a file
 104                         http.Error(w, "object name conflicts with existing object", http.StatusBadRequest)
 105                         return true
 106                 }
 107                 f, err := fs.OpenFile(fspath, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
 108                 if os.IsNotExist(err) {
 109                         // create missing intermediate directories, then try again
 110                         for i, c := range fspath {
 111                                 if i > 0 && c == '/' {
 112                                         dir := fspath[:i]
 113                                         if strings.HasSuffix(dir, "/") {
 114                                                 err = errors.New("invalid object name (consecutive '/' chars)")
 115                                                 http.Error(w, err.Error(), http.StatusBadRequest)
 116                                                 return true
 117                                         }
 118                                         err := fs.Mkdir(dir, 0755)
 119                                         if err != nil && err != os.ErrExist {
 120                                                 err = fmt.Errorf("mkdir %q failed: %w", dir, err)
 121                                                 http.Error(w, err.Error(), http.StatusInternalServerError)
 122                                                 return true
 123                                         }
 124                                 }
 125                         }
 126                         f, err = fs.OpenFile(fspath, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
 127                 }
 128                 if err != nil {
 129                         err = fmt.Errorf("open %q failed: %w", r.URL.Path, err)
 130                         http.Error(w, err.Error(), http.StatusBadRequest)
 131                         return true
 132                 }
 133                 defer f.Close()
 134                 _, err = io.Copy(f, r.Body)
 135                 if err != nil {
 136                         err = fmt.Errorf("write to %q failed: %w", r.URL.Path, err)
 137                         http.Error(w, err.Error(), http.StatusBadGateway)
 138                         return true
 139                 }
 140                 err = f.Close()
 141                 if err != nil {
 142                         err = fmt.Errorf("write to %q failed: close: %w", r.URL.Path, err)
 143                         http.Error(w, err.Error(), http.StatusBadGateway)
 144                         return true
 145                 }
 146                 err = fs.Sync()
 147                 if err != nil {
 148                         err = fmt.Errorf("sync failed: %w", err)
 149                         http.Error(w, err.Error(), http.StatusInternalServerError)
 150                         return true
 151                 }
 152                 w.WriteHeader(http.StatusOK)
 153                 return true
 154         default:
 155                 http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
 156                 return true
 157         }
 158 }
 159
 160 func walkFS(fs arvados.CustomFileSystem, path string, ignoreNotFound bool, fn func(path string, fi os.FileInfo) error) error {
 161         f, err := fs.Open(path)
 162         if os.IsNotExist(err) && ignoreNotFound {
 163                 return nil
 164         } else if err != nil {
 165                 return fmt.Errorf("open %q: %w", path, err)
 166         }
 167         defer f.Close()
 168         if path == "/" {
 169                 path = ""
 170         }
 171         fis, err := f.Readdir(-1)
 172         if err != nil {
 173                 return err
 174         }
 175         sort.Slice(fis, func(i, j int) bool { return fis[i].Name() < fis[j].Name() })
 176         for _, fi := range fis {
 177                 err = fn(path+"/"+fi.Name(), fi)
 178                 if err == filepath.SkipDir {
 179                         continue
 180                 } else if err != nil {
 181                         return err
 182                 }
 183                 if fi.IsDir() {
 184                         err = walkFS(fs, path+"/"+fi.Name(), false, fn)
 185                         if err != nil {
 186                                 return err
 187                         }
 188                 }
 189         }
 190         return nil
 191 }
 192
 193 var errDone = errors.New("done")
 194
 195 func (h *handler) s3list(w http.ResponseWriter, r *http.Request, fs arvados.CustomFileSystem) {
 196         var params struct {
 197                 bucket    string
 198                 delimiter string
 199                 marker    string
 200                 maxKeys   int
 201                 prefix    string
 202         }
 203         params.bucket = strings.SplitN(r.URL.Path[1:], "/", 2)[0]
 204         params.delimiter = r.FormValue("delimiter")
 205         params.marker = r.FormValue("marker")
 206         if mk, _ := strconv.ParseInt(r.FormValue("max-keys"), 10, 64); mk > 0 && mk < s3MaxKeys {
 207                 params.maxKeys = int(mk)
 208         } else {
 209                 params.maxKeys = s3MaxKeys
 210         }
 211         params.prefix = r.FormValue("prefix")
 212
 213         bucketdir := "by_id/" + params.bucket
 214         // walkpath is the directory (relative to bucketdir) we need
 215         // to walk: the innermost directory that is guaranteed to
 216         // contain all paths that have the requested prefix. Examples:
 217         // prefix "foo/bar"  => walkpath "foo"
 218         // prefix "foo/bar/" => walkpath "foo/bar"
 219         // prefix "foo"      => walkpath ""
 220         // prefix ""         => walkpath ""
 221         walkpath := params.prefix
 222         if cut := strings.LastIndex(walkpath, "/"); cut >= 0 {
 223                 walkpath = walkpath[:cut]
 224         } else {
 225                 walkpath = ""
 226         }
 227
 228         resp := s3.ListResp{
 229                 Name:      strings.SplitN(r.URL.Path[1:], "/", 2)[0],
 230                 Prefix:    params.prefix,
 231                 Delimiter: params.delimiter,
 232                 Marker:    params.marker,
 233                 MaxKeys:   params.maxKeys,
 234         }
 235         commonPrefixes := map[string]bool{}
 236         err := walkFS(fs, strings.TrimSuffix(bucketdir+"/"+walkpath, "/"), true, func(path string, fi os.FileInfo) error {
 237                 path = path[len(bucketdir)+1:]
 238                 if len(path) <= len(params.prefix) {
 239                         if path > params.prefix[:len(path)] {
 240                                 // with prefix "foobar", walking "fooz" means we're done
 241                                 return errDone
 242                         }
 243                         if path < params.prefix[:len(path)] {
 244                                 // with prefix "foobar", walking "foobag" is pointless
 245                                 return filepath.SkipDir
 246                         }
 247                         if fi.IsDir() && !strings.HasPrefix(params.prefix+"/", path+"/") {
 248                                 // with prefix "foo/bar", walking "fo"
 249                                 // is pointless (but walking "foo" or
 250                                 // "foo/bar" is necessary)
 251                                 return filepath.SkipDir
 252                         }
 253                         if len(path) < len(params.prefix) {
 254                                 // can't skip anything, and this entry
 255                                 // isn't in the results, so just
 256                                 // continue descent
 257                                 return nil
 258                         }
 259                 } else {
 260                         if path[:len(params.prefix)] > params.prefix {
 261                                 // with prefix "foobar", nothing we
 262                                 // see after "foozzz" is relevant
 263                                 return errDone
 264                         }
 265                 }
 266                 if path < params.marker || path < params.prefix {
 267                         return nil
 268                 }
 269                 if fi.IsDir() {
 270                         return nil
 271                 }
 272                 if params.delimiter != "" {
 273                         idx := strings.Index(path[len(params.prefix):], params.delimiter)
 274                         if idx >= 0 {
 275                                 // with prefix "foobar" and delimiter
 276                                 // "z", when we hit "foobar/baz", we
 277                                 // add "/baz" to commonPrefixes and
 278                                 // stop descending (note that even if
 279                                 // delimiter is "/" we don't add
 280                                 // anything to commonPrefixes when
 281                                 // seeing a dir: we wait until we see
 282                                 // a file, so we don't incorrectly
 283                                 // return results for empty dirs)
 284                                 commonPrefixes[path[:len(params.prefix)+idx+1]] = true
 285                                 return filepath.SkipDir
 286                         }
 287                 }
 288                 if len(resp.Contents)+len(commonPrefixes) >= params.maxKeys {
 289                         resp.IsTruncated = true
 290                         if params.delimiter != "" {
 291                                 resp.NextMarker = path
 292                         }
 293                         return errDone
 294                 }
 295                 resp.Contents = append(resp.Contents, s3.Key{
 296                         Key: path,
 297                 })
 298                 return nil
 299         })
 300         if err != nil && err != errDone {
 301                 http.Error(w, err.Error(), http.StatusInternalServerError)
 302                 return
 303         }
 304         if params.delimiter != "" {
 305                 for prefix := range commonPrefixes {
 306                         resp.CommonPrefixes = append(resp.CommonPrefixes, prefix)
 307                         sort.Strings(resp.CommonPrefixes)
 308                 }
 309         }
 310         wrappedResp := struct {
 311                 XMLName string `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListBucketResult"`
 312                 s3.ListResp
 313         }{"", resp}
 314         w.Header().Set("Content-Type", "application/xml")
 315         io.WriteString(w, xml.Header)
 316         if err := xml.NewEncoder(w).Encode(wrappedResp); err != nil {
 317                 ctxlog.FromContext(r.Context()).WithError(err).Error("error writing xml response")
 318         }
 319 }