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         "git.arvados.org/arvados.git/tmp/GOPATH/src/github.com/AdRoll/goamz/s3"
  22 )
  23
  24 // serveS3 handles r and returns true if r is a request from an S3
  25 // client, otherwise it returns false.
  26 func (h *handler) serveS3(w http.ResponseWriter, r *http.Request) bool {
  27         var token string
  28         if auth := r.Header.Get("Authorization"); strings.HasPrefix(auth, "AWS ") {
  29                 split := strings.SplitN(auth[4:], ":", 2)
  30                 if len(split) < 2 {
  31                         w.WriteHeader(http.StatusUnauthorized)
  32                         return true
  33                 }
  34                 token = split[0]
  35         } else if strings.HasPrefix(auth, "AWS4-HMAC-SHA256 ") {
  36                 w.WriteHeader(http.StatusBadRequest)
  37                 fmt.Println(w, "V4 signature is not supported")
  38                 return true
  39         } else {
  40                 return false
  41         }
  42
  43         _, kc, client, release, err := h.getClients(r.Header.Get("X-Request-Id"), token)
  44         if err != nil {
  45                 http.Error(w, "Pool failed: "+h.clientPool.Err().Error(), http.StatusInternalServerError)
  46                 return true
  47         }
  48         defer release()
  49
  50         fs := client.SiteFileSystem(kc)
  51         fs.ForwardSlashNameSubstitution(h.Config.cluster.Collections.ForwardSlashNameSubstitution)
  52
  53         switch {
  54         case r.Method == "GET" && strings.Count(strings.TrimSuffix(r.URL.Path, "/"), "/") == 1:
  55                 // Path is "/{uuid}" or "/{uuid}/", has no object name
  56                 h.s3list(w, r, fs)
  57                 return true
  58         case r.Method == "GET":
  59                 fspath := "/by_id" + r.URL.Path
  60                 fi, err := fs.Stat(fspath)
  61                 if os.IsNotExist(err) ||
  62                         (err != nil && err.Error() == "not a directory") ||
  63                         (fi != nil && fi.IsDir()) {
  64                         http.Error(w, "not found", http.StatusNotFound)
  65                         return true
  66                 }
  67                 // shallow copy r, and change URL path
  68                 r := *r
  69                 r.URL.Path = fspath
  70                 http.FileServer(fs).ServeHTTP(w, &r)
  71                 return true
  72         case r.Method == "PUT":
  73                 if strings.HasSuffix(r.URL.Path, "/") {
  74                         http.Error(w, "invalid object name (trailing '/' char)", http.StatusBadRequest)
  75                         return true
  76                 }
  77                 fspath := "by_id" + r.URL.Path
  78                 _, err = fs.Stat(fspath)
  79                 if err != nil && err.Error() == "not a directory" {
  80                         // requested foo/bar, but foo is a file
  81                         http.Error(w, "object name conflicts with existing object", http.StatusBadRequest)
  82                         return true
  83                 }
  84                 f, err := fs.OpenFile(fspath, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
  85                 if os.IsNotExist(err) {
  86                         // create missing intermediate directories, then try again
  87                         for i, c := range fspath {
  88                                 if i > 0 && c == '/' {
  89                                         dir := fspath[:i]
  90                                         if strings.HasSuffix(dir, "/") {
  91                                                 err = errors.New("invalid object name (consecutive '/' chars)")
  92                                                 http.Error(w, err.Error(), http.StatusBadRequest)
  93                                                 return true
  94                                         }
  95                                         err := fs.Mkdir(dir, 0755)
  96                                         if err != nil && err != os.ErrExist {
  97                                                 err = fmt.Errorf("mkdir %q failed: %w", dir, err)
  98                                                 http.Error(w, err.Error(), http.StatusInternalServerError)
  99                                                 return true
 100                                         }
 101                                 }
 102                         }
 103                         f, err = fs.OpenFile(fspath, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
 104                 }
 105                 if err != nil {
 106                         err = fmt.Errorf("open %q failed: %w", r.URL.Path, err)
 107                         http.Error(w, err.Error(), http.StatusBadRequest)
 108                         return true
 109                 }
 110                 defer f.Close()
 111                 _, err = io.Copy(f, r.Body)
 112                 if err != nil {
 113                         err = fmt.Errorf("write to %q failed: %w", r.URL.Path, err)
 114                         http.Error(w, err.Error(), http.StatusBadGateway)
 115                         return true
 116                 }
 117                 err = f.Close()
 118                 if err != nil {
 119                         err = fmt.Errorf("write to %q failed: %w", r.URL.Path, err)
 120                         http.Error(w, err.Error(), http.StatusBadGateway)
 121                         return true
 122                 }
 123                 err = fs.Sync()
 124                 if err != nil {
 125                         err = fmt.Errorf("sync failed: %w", err)
 126                         http.Error(w, err.Error(), http.StatusInternalServerError)
 127                         return true
 128                 }
 129                 w.WriteHeader(http.StatusOK)
 130                 return true
 131         default:
 132                 http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
 133                 return true
 134         }
 135 }
 136
 137 func walkFS(fs arvados.CustomFileSystem, path string, fn func(path string, fi os.FileInfo) error) error {
 138         f, err := fs.Open(path)
 139         if err != nil {
 140                 return fmt.Errorf("open %q: %w", path, err)
 141         }
 142         defer f.Close()
 143         if path == "/" {
 144                 path = ""
 145         }
 146         fis, err := f.Readdir(-1)
 147         if err != nil {
 148                 return err
 149         }
 150         sort.Slice(fis, func(i, j int) bool { return fis[i].Name() < fis[j].Name() })
 151         for _, fi := range fis {
 152                 err = fn(path+"/"+fi.Name(), fi)
 153                 if err == filepath.SkipDir {
 154                         continue
 155                 } else if err != nil {
 156                         return err
 157                 }
 158                 if fi.IsDir() {
 159                         err = walkFS(fs, path+"/"+fi.Name(), fn)
 160                         if err != nil {
 161                                 return err
 162                         }
 163                 }
 164         }
 165         return nil
 166 }
 167
 168 var errDone = errors.New("done")
 169
 170 func (h *handler) s3list(w http.ResponseWriter, r *http.Request, fs arvados.CustomFileSystem) {
 171         var params struct {
 172                 bucket    string
 173                 delimiter string
 174                 marker    string
 175                 maxKeys   int
 176                 prefix    string
 177         }
 178         params.bucket = strings.SplitN(r.URL.Path[1:], "/", 2)[0]
 179         params.delimiter = r.FormValue("delimiter")
 180         params.marker = r.FormValue("marker")
 181         if mk, _ := strconv.ParseInt(r.FormValue("max-keys"), 10, 64); mk > 0 {
 182                 params.maxKeys = int(mk)
 183         } else {
 184                 params.maxKeys = 100
 185         }
 186         params.prefix = r.FormValue("prefix")
 187
 188         bucketdir := "by_id/" + params.bucket
 189         // walkpath is the directory (relative to bucketdir) we need
 190         // to walk: the innermost directory that is guaranteed to
 191         // contain all paths that have the requested prefix. Examples:
 192         // prefix "foo/bar"  => walkpath "foo"
 193         // prefix "foo/bar/" => walkpath "foo/bar"
 194         // prefix "foo"      => walkpath ""
 195         // prefix ""         => walkpath ""
 196         walkpath := params.prefix
 197         if !strings.HasSuffix(walkpath, "/") {
 198                 walkpath, _ = filepath.Split(walkpath)
 199         }
 200         walkpath = strings.TrimSuffix(walkpath, "/")
 201
 202         type commonPrefix struct {
 203                 Prefix string
 204         }
 205         type serverListResponse struct {
 206                 s3.ListResp
 207                 CommonPrefixes []commonPrefix
 208         }
 209         resp := serverListResponse{ListResp: s3.ListResp{
 210                 Name:      strings.SplitN(r.URL.Path[1:], "/", 2)[0],
 211                 Prefix:    params.prefix,
 212                 Delimiter: params.delimiter,
 213                 Marker:    params.marker,
 214                 MaxKeys:   params.maxKeys,
 215         }}
 216         err := walkFS(fs, strings.TrimSuffix(bucketdir+"/"+walkpath, "/"), func(path string, fi os.FileInfo) error {
 217                 path = path[len(bucketdir)+1:]
 218                 if !strings.HasPrefix(path, params.prefix) {
 219                         return filepath.SkipDir
 220                 }
 221                 if fi.IsDir() {
 222                         return nil
 223                 }
 224                 if path < params.marker {
 225                         return nil
 226                 }
 227                 // TODO: check delimiter, roll up common prefixes
 228                 if len(resp.Contents)+len(resp.CommonPrefixes) >= params.maxKeys {
 229                         resp.IsTruncated = true
 230                         if params.delimiter == "" {
 231                                 resp.NextMarker = path
 232                         }
 233                         return errDone
 234                 }
 235                 resp.ListResp.Contents = append(resp.ListResp.Contents, s3.Key{
 236                         Key: path,
 237                 })
 238                 return nil
 239         })
 240         if err != nil && err != errDone {
 241                 http.Error(w, err.Error(), http.StatusInternalServerError)
 242                 return
 243         }
 244         if err := xml.NewEncoder(w).Encode(resp); err != nil {
 245                 ctxlog.FromContext(r.Context()).WithError(err).Error("error writing xml response")
 246         }
 247 }