1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
|
package weed_server
import (
"encoding/base64"
"encoding/hex"
"errors"
"fmt"
"io"
"math"
"mime"
"net/http"
"path/filepath"
"strconv"
"strings"
"time"
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
"github.com/seaweedfs/seaweedfs/weed/security"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/stats"
"github.com/seaweedfs/seaweedfs/weed/util"
)
// Validates the preconditions. Returns true if GET/HEAD operation should not proceed.
// Preconditions supported are:
//
// If-Modified-Since
// If-Unmodified-Since
// If-Match
// If-None-Match
func checkPreconditions(w http.ResponseWriter, r *http.Request, entry *filer.Entry) bool {
etag := filer.ETagEntry(entry)
/// When more than one conditional request header field is present in a
/// request, the order in which the fields are evaluated becomes
/// important. In practice, the fields defined in this document are
/// consistently implemented in a single, logical order, since "lost
/// update" preconditions have more strict requirements than cache
/// validation, a validated cache is more efficient than a partial
/// response, and entity tags are presumed to be more accurate than date
/// validators. https://tools.ietf.org/html/rfc7232#section-5
if entry.Attr.Mtime.IsZero() {
return false
}
w.Header().Set("Last-Modified", entry.Attr.Mtime.UTC().Format(http.TimeFormat))
ifMatchETagHeader := r.Header.Get("If-Match")
ifUnmodifiedSinceHeader := r.Header.Get("If-Unmodified-Since")
if ifMatchETagHeader != "" {
if util.CanonicalizeETag(etag) != util.CanonicalizeETag(ifMatchETagHeader) {
w.WriteHeader(http.StatusPreconditionFailed)
return true
}
} else if ifUnmodifiedSinceHeader != "" {
if t, parseError := time.Parse(http.TimeFormat, ifUnmodifiedSinceHeader); parseError == nil {
if t.Before(entry.Attr.Mtime) {
w.WriteHeader(http.StatusPreconditionFailed)
return true
}
}
}
ifNoneMatchETagHeader := r.Header.Get("If-None-Match")
ifModifiedSinceHeader := r.Header.Get("If-Modified-Since")
if ifNoneMatchETagHeader != "" {
if util.CanonicalizeETag(etag) == util.CanonicalizeETag(ifNoneMatchETagHeader) {
SetEtag(w, etag)
w.WriteHeader(http.StatusNotModified)
return true
}
} else if ifModifiedSinceHeader != "" {
if t, parseError := time.Parse(http.TimeFormat, ifModifiedSinceHeader); parseError == nil {
if !t.Before(entry.Attr.Mtime) {
SetEtag(w, etag)
w.WriteHeader(http.StatusNotModified)
return true
}
}
}
return false
}
func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
path := r.URL.Path
isForDirectory := strings.HasSuffix(path, "/")
if isForDirectory && len(path) > 1 {
path = path[:len(path)-1]
}
entry, err := fs.filer.FindEntry(ctx, util.FullPath(path))
if err != nil {
if path == "/" {
fs.listDirectoryHandler(w, r)
return
}
if err == filer_pb.ErrNotFound {
glog.V(2).InfofCtx(ctx, "Not found %s: %v", path, err)
stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadNotFound).Inc()
w.WriteHeader(http.StatusNotFound)
} else {
glog.ErrorfCtx(ctx, "Internal %s: %v", path, err)
stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadInternal).Inc()
w.WriteHeader(http.StatusInternalServerError)
}
return
}
query := r.URL.Query()
if entry.IsDirectory() {
if fs.option.DisableDirListing {
w.WriteHeader(http.StatusForbidden)
return
}
if query.Get("metadata") == "true" {
writeJsonQuiet(w, r, http.StatusOK, entry)
return
}
if entry.Attr.Mime == "" || (entry.Attr.Mime == s3_constants.FolderMimeType && r.Header.Get(s3_constants.AmzIdentityId) == "") {
// Don't return directory meta if config value is set to true
if fs.option.ExposeDirectoryData == false {
writeJsonError(w, r, http.StatusForbidden, errors.New("directory listing is disabled"))
return
}
// return index of directory for non s3 gateway
fs.listDirectoryHandler(w, r)
return
}
// inform S3 API this is a user created directory key object
w.Header().Set(s3_constants.SeaweedFSIsDirectoryKey, "true")
}
if isForDirectory && entry.Attr.Mime != s3_constants.FolderMimeType {
w.WriteHeader(http.StatusNotFound)
return
}
if query.Get("metadata") == "true" {
if query.Get("resolveManifest") == "true" {
if entry.Chunks, _, err = filer.ResolveChunkManifest(
ctx,
fs.filer.MasterClient.GetLookupFileIdFunction(),
entry.GetChunks(), 0, math.MaxInt64); err != nil {
err = fmt.Errorf("failed to resolve chunk manifest, err: %s", err.Error())
writeJsonError(w, r, http.StatusInternalServerError, err)
return
}
}
writeJsonQuiet(w, r, http.StatusOK, entry)
return
}
if checkPreconditions(w, r, entry) {
return
}
var etag string
if partNumber, errNum := strconv.Atoi(r.Header.Get(s3_constants.SeaweedFSPartNumber)); errNum == nil {
if len(entry.Chunks) < partNumber {
stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadChunk).Inc()
w.WriteHeader(http.StatusBadRequest)
w.Write([]byte("InvalidPart"))
return
}
w.Header().Set(s3_constants.AmzMpPartsCount, strconv.Itoa(len(entry.Chunks)))
partChunk := entry.GetChunks()[partNumber-1]
md5, _ := base64.StdEncoding.DecodeString(partChunk.ETag)
etag = hex.EncodeToString(md5)
r.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", partChunk.Offset, uint64(partChunk.Offset)+partChunk.Size-1))
} else {
etag = filer.ETagEntry(entry)
}
w.Header().Set("Accept-Ranges", "bytes")
// mime type
mimeType := entry.Attr.Mime
if mimeType == "" {
if ext := filepath.Ext(entry.Name()); ext != "" {
mimeType = mime.TypeByExtension(ext)
}
}
if mimeType != "" {
w.Header().Set("Content-Type", mimeType)
} else {
w.Header().Set("Content-Type", "application/octet-stream")
}
// print out the header from extended properties
for k, v := range entry.Extended {
if !strings.HasPrefix(k, "xattr-") && !s3_constants.IsSeaweedFSInternalHeader(k) {
// "xattr-" prefix is set in filesys.XATTR_PREFIX
// IsSeaweedFSInternalHeader filters internal metadata that should not become HTTP headers
w.Header().Set(k, string(v))
}
}
//Seaweed custom header are not visible to Vue or javascript
seaweedHeaders := []string{}
for header := range w.Header() {
if strings.HasPrefix(header, "Seaweed-") {
seaweedHeaders = append(seaweedHeaders, header)
}
}
seaweedHeaders = append(seaweedHeaders, "Content-Disposition")
w.Header().Set("Access-Control-Expose-Headers", strings.Join(seaweedHeaders, ","))
//set tag count
tagCount := 0
for k := range entry.Extended {
if strings.HasPrefix(k, s3_constants.AmzObjectTagging+"-") {
tagCount++
}
}
if tagCount > 0 {
w.Header().Set(s3_constants.AmzTagCount, strconv.Itoa(tagCount))
}
// Set SSE metadata headers for S3 API consumption
if sseIV, exists := entry.Extended[s3_constants.SeaweedFSSSEIV]; exists {
// Convert binary IV to base64 for HTTP header
ivBase64 := base64.StdEncoding.EncodeToString(sseIV)
w.Header().Set(s3_constants.SeaweedFSSSEIVHeader, ivBase64)
}
// Set SSE-C algorithm and key MD5 headers for S3 API response
if sseAlgorithm, exists := entry.Extended[s3_constants.AmzServerSideEncryptionCustomerAlgorithm]; exists {
w.Header().Set(s3_constants.AmzServerSideEncryptionCustomerAlgorithm, string(sseAlgorithm))
}
if sseKeyMD5, exists := entry.Extended[s3_constants.AmzServerSideEncryptionCustomerKeyMD5]; exists {
w.Header().Set(s3_constants.AmzServerSideEncryptionCustomerKeyMD5, string(sseKeyMD5))
}
if sseKMSKey, exists := entry.Extended[s3_constants.SeaweedFSSSEKMSKey]; exists {
// Convert binary KMS metadata to base64 for HTTP header
kmsBase64 := base64.StdEncoding.EncodeToString(sseKMSKey)
w.Header().Set(s3_constants.SeaweedFSSSEKMSKeyHeader, kmsBase64)
}
if _, exists := entry.Extended[s3_constants.SeaweedFSSSES3Key]; exists {
// Set standard S3 SSE-S3 response header (not the internal SeaweedFS header)
w.Header().Set(s3_constants.AmzServerSideEncryption, s3_constants.SSEAlgorithmAES256)
}
SetEtag(w, etag)
filename := entry.Name()
AdjustPassthroughHeaders(w, r, filename)
// For range processing, use the original content size, not the encrypted size
// entry.Size() returns max(chunk_sizes, file_size) where chunk_sizes include encryption overhead
// For SSE objects, we need the original unencrypted size for proper range validation
totalSize := int64(entry.FileSize)
if r.Method == http.MethodHead {
w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
return
}
ProcessRangeRequest(r, w, totalSize, mimeType, func(offset int64, size int64) (filer.DoStreamContent, error) {
if offset+size <= int64(len(entry.Content)) {
return func(writer io.Writer) error {
_, err := writer.Write(entry.Content[offset : offset+size])
if err != nil {
stats.FilerHandlerCounter.WithLabelValues(stats.ErrorWriteEntry).Inc()
glog.ErrorfCtx(ctx, "failed to write entry content: %v", err)
}
return err
}, nil
}
chunks := entry.GetChunks()
if entry.IsInRemoteOnly() {
dir, name := entry.FullPath.DirAndName()
if resp, err := fs.CacheRemoteObjectToLocalCluster(ctx, &filer_pb.CacheRemoteObjectToLocalClusterRequest{
Directory: dir,
Name: name,
}); err != nil {
stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadCache).Inc()
glog.ErrorfCtx(ctx, "CacheRemoteObjectToLocalCluster %s: %v", entry.FullPath, err)
return nil, fmt.Errorf("cache %s: %v", entry.FullPath, err)
} else {
chunks = resp.Entry.GetChunks()
}
}
streamFn, err := filer.PrepareStreamContentWithThrottler(ctx, fs.filer.MasterClient, fs.maybeGetVolumeReadJwtAuthorizationToken, chunks, offset, size, fs.option.DownloadMaxBytesPs)
if err != nil {
stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
glog.ErrorfCtx(ctx, "failed to prepare stream content %s: %v", r.URL, err)
return nil, err
}
return func(writer io.Writer) error {
err := streamFn(writer)
if err != nil {
stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
glog.ErrorfCtx(ctx, "failed to stream content %s: %v", r.URL, err)
}
return err
}, nil
})
}
func (fs *FilerServer) maybeGetVolumeReadJwtAuthorizationToken(fileId string) string {
return string(security.GenJwtForVolumeServer(fs.volumeGuard.ReadSigningKey, fs.volumeGuard.ReadExpiresAfterSec, fileId))
}
|