mirror of
https://github.com/minio/minio.git
synced 2025-04-19 02:05:24 -04:00
Change SelectAPI to use new GetObjectNInfo API (#6373)
This PR also removes some double checks
This commit is contained in:
parent
a13cd7b7c4
commit
5b05df215a
@ -73,10 +73,6 @@ func setHeadGetRespHeaders(w http.ResponseWriter, reqParams url.Values) {
|
|||||||
// also specify a data serialization format (JSON, CSV) of the object.
|
// also specify a data serialization format (JSON, CSV) of the object.
|
||||||
func (api objectAPIHandlers) SelectObjectContentHandler(w http.ResponseWriter, r *http.Request) {
|
func (api objectAPIHandlers) SelectObjectContentHandler(w http.ResponseWriter, r *http.Request) {
|
||||||
ctx := newContext(r, w, "SelectObject")
|
ctx := newContext(r, w, "SelectObject")
|
||||||
var object, bucket string
|
|
||||||
vars := mux.Vars(r)
|
|
||||||
bucket = vars["bucket"]
|
|
||||||
object = vars["object"]
|
|
||||||
|
|
||||||
// Fetch object stat info.
|
// Fetch object stat info.
|
||||||
objectAPI := api.ObjectAPI()
|
objectAPI := api.ObjectAPI()
|
||||||
@ -85,27 +81,43 @@ func (api objectAPIHandlers) SelectObjectContentHandler(w http.ResponseWriter, r
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
getObjectInfo := objectAPI.GetObjectInfo
|
if crypto.S3.IsRequested(r.Header) || crypto.S3KMS.IsRequested(r.Header) { // If SSE-S3 or SSE-KMS present -> AWS fails with undefined error
|
||||||
if api.CacheAPI() != nil {
|
writeErrorResponse(w, ErrBadRequest, r.URL)
|
||||||
getObjectInfo = api.CacheAPI().GetObjectInfo
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
vars := mux.Vars(r)
|
||||||
|
bucket := vars["bucket"]
|
||||||
|
object := vars["object"]
|
||||||
|
|
||||||
|
// Check for auth type to return S3 compatible error.
|
||||||
|
// type to return the correct error (NoSuchKey vs AccessDenied)
|
||||||
if s3Error := checkRequestAuthType(ctx, r, policy.GetObjectAction, bucket, object); s3Error != ErrNone {
|
if s3Error := checkRequestAuthType(ctx, r, policy.GetObjectAction, bucket, object); s3Error != ErrNone {
|
||||||
if getRequestAuthType(r) == authTypeAnonymous {
|
if getRequestAuthType(r) == authTypeAnonymous {
|
||||||
// As per "Permission" section in
|
// As per "Permission" section in
|
||||||
// https://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectGET.html If
|
// https://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectGET.html
|
||||||
// the object you request does not exist, the error Amazon S3 returns
|
// If the object you request does not exist,
|
||||||
// depends on whether you also have the s3:ListBucket permission. * If you
|
// the error Amazon S3 returns depends on
|
||||||
// have the s3:ListBucket permission on the bucket, Amazon S3 will return
|
// whether you also have the s3:ListBucket
|
||||||
// an HTTP status code 404 ("no such key") error. * if you don’t have the
|
// permission.
|
||||||
// s3:ListBucket permission, Amazon S3 will return an HTTP status code 403
|
// * If you have the s3:ListBucket permission
|
||||||
// ("access denied") error.`
|
// on the bucket, Amazon S3 will return an
|
||||||
|
// HTTP status code 404 ("no such key")
|
||||||
|
// error.
|
||||||
|
// * if you don’t have the s3:ListBucket
|
||||||
|
// permission, Amazon S3 will return an HTTP
|
||||||
|
// status code 403 ("access denied") error.`
|
||||||
if globalPolicySys.IsAllowed(policy.Args{
|
if globalPolicySys.IsAllowed(policy.Args{
|
||||||
Action: policy.ListBucketAction,
|
Action: policy.ListBucketAction,
|
||||||
BucketName: bucket,
|
BucketName: bucket,
|
||||||
ConditionValues: getConditionValues(r, ""),
|
ConditionValues: getConditionValues(r, ""),
|
||||||
IsOwner: false,
|
IsOwner: false,
|
||||||
}) {
|
}) {
|
||||||
|
getObjectInfo := objectAPI.GetObjectInfo
|
||||||
|
if api.CacheAPI() != nil {
|
||||||
|
getObjectInfo = api.CacheAPI().GetObjectInfo
|
||||||
|
}
|
||||||
|
|
||||||
_, err := getObjectInfo(ctx, bucket, object)
|
_, err := getObjectInfo(ctx, bucket, object)
|
||||||
if toAPIErrorCode(err) == ErrNoSuchKey {
|
if toAPIErrorCode(err) == ErrNoSuchKey {
|
||||||
s3Error = ErrNoSuchKey
|
s3Error = ErrNoSuchKey
|
||||||
@ -115,25 +127,55 @@ func (api objectAPIHandlers) SelectObjectContentHandler(w http.ResponseWriter, r
|
|||||||
writeErrorResponse(w, s3Error, r.URL)
|
writeErrorResponse(w, s3Error, r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Get request range.
|
||||||
|
rangeHeader := r.Header.Get("Range")
|
||||||
|
if rangeHeader != "" {
|
||||||
|
writeErrorResponse(w, ErrUnsupportedRangeHeader, r.URL)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
if r.ContentLength <= 0 {
|
if r.ContentLength <= 0 {
|
||||||
writeErrorResponse(w, ErrEmptyRequestBody, r.URL)
|
writeErrorResponse(w, ErrEmptyRequestBody, r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
var selectReq ObjectSelectRequest
|
var selectReq ObjectSelectRequest
|
||||||
if err := xmlDecoder(r.Body, &selectReq, r.ContentLength); err != nil {
|
if err := xmlDecoder(r.Body, &selectReq, r.ContentLength); err != nil {
|
||||||
writeErrorResponse(w, ErrMalformedXML, r.URL)
|
writeErrorResponse(w, ErrMalformedXML, r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
objInfo, err := getObjectInfo(ctx, bucket, object)
|
if !strings.EqualFold(string(selectReq.ExpressionType), "SQL") {
|
||||||
if err != nil {
|
writeErrorResponse(w, ErrInvalidExpressionType, r.URL)
|
||||||
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// Get request range.
|
if len(selectReq.Expression) >= s3select.MaxExpressionLength {
|
||||||
rangeHeader := r.Header.Get("Range")
|
writeErrorResponse(w, ErrExpressionTooLong, r.URL)
|
||||||
if rangeHeader != "" {
|
return
|
||||||
writeErrorResponse(w, ErrUnsupportedRangeHeader, r.URL)
|
}
|
||||||
|
if selectReq.InputSerialization.CSV.FileHeaderInfo != CSVFileHeaderInfoUse &&
|
||||||
|
selectReq.InputSerialization.CSV.FileHeaderInfo != CSVFileHeaderInfoNone &&
|
||||||
|
selectReq.InputSerialization.CSV.FileHeaderInfo != CSVFileHeaderInfoIgnore &&
|
||||||
|
selectReq.InputSerialization.CSV.FileHeaderInfo != "" {
|
||||||
|
writeErrorResponse(w, ErrInvalidFileHeaderInfo, r.URL)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if selectReq.OutputSerialization.CSV.QuoteFields != CSVQuoteFieldsAlways &&
|
||||||
|
selectReq.OutputSerialization.CSV.QuoteFields != CSVQuoteFieldsAsNeeded &&
|
||||||
|
selectReq.OutputSerialization.CSV.QuoteFields != "" {
|
||||||
|
writeErrorResponse(w, ErrInvalidQuoteFields, r.URL)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
getObjectNInfo := objectAPI.GetObjectNInfo
|
||||||
|
if api.CacheAPI() != nil {
|
||||||
|
getObjectNInfo = api.CacheAPI().GetObjectNInfo
|
||||||
|
}
|
||||||
|
|
||||||
|
objInfo, reader, err := getObjectNInfo(ctx, bucket, object, nil)
|
||||||
|
if err != nil {
|
||||||
|
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -157,63 +199,48 @@ func (api objectAPIHandlers) SelectObjectContentHandler(w http.ResponseWriter, r
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if !strings.EqualFold(string(selectReq.ExpressionType), "SQL") {
|
|
||||||
writeErrorResponse(w, ErrInvalidExpressionType, r.URL)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if len(selectReq.Expression) >= s3select.MaxExpressionLength {
|
|
||||||
writeErrorResponse(w, ErrExpressionTooLong, r.URL)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if selectReq.InputSerialization.CSV.FileHeaderInfo != CSVFileHeaderInfoUse &&
|
|
||||||
selectReq.InputSerialization.CSV.FileHeaderInfo != CSVFileHeaderInfoNone &&
|
|
||||||
selectReq.InputSerialization.CSV.FileHeaderInfo != CSVFileHeaderInfoIgnore &&
|
|
||||||
selectReq.InputSerialization.CSV.FileHeaderInfo != "" {
|
|
||||||
writeErrorResponse(w, ErrInvalidFileHeaderInfo, r.URL)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if selectReq.OutputSerialization.CSV.QuoteFields != CSVQuoteFieldsAlways &&
|
|
||||||
selectReq.OutputSerialization.CSV.QuoteFields != CSVQuoteFieldsAsNeeded &&
|
|
||||||
selectReq.OutputSerialization.CSV.QuoteFields != "" {
|
|
||||||
writeErrorResponse(w, ErrInvalidQuoteFields, r.URL)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
getObject := objectAPI.GetObject
|
// If object is encrypted, we avoid the cache layer.
|
||||||
if api.CacheAPI() != nil && !crypto.SSEC.IsRequested(r.Header) {
|
if crypto.IsEncrypted(objInfo.UserDefined) && api.CacheAPI() != nil {
|
||||||
getObject = api.CacheAPI().GetObject
|
// Close the existing reader before re-querying the backend
|
||||||
|
if reader != nil {
|
||||||
|
reader.Close()
|
||||||
}
|
}
|
||||||
|
// Query the backend
|
||||||
reader, pipewriter := io.Pipe()
|
objInfo, reader, err = objectAPI.GetObjectNInfo(ctx, bucket, object, nil)
|
||||||
|
|
||||||
// Get the object.
|
|
||||||
var startOffset int64
|
|
||||||
length := objInfo.Size
|
|
||||||
|
|
||||||
var writer io.Writer
|
|
||||||
writer = pipewriter
|
|
||||||
if objectAPI.IsEncryptionSupported() {
|
|
||||||
if crypto.SSEC.IsRequested(r.Header) {
|
|
||||||
// Response writer should be limited early on for decryption upto required length,
|
|
||||||
// additionally also skipping mod(offset)64KiB boundaries.
|
|
||||||
writer = ioutil.LimitedWriter(writer, startOffset%(64*1024), length)
|
|
||||||
|
|
||||||
writer, startOffset, length, err = DecryptBlocksRequest(writer, r, bucket,
|
|
||||||
object, startOffset, length, objInfo, false)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
|
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
go func() {
|
|
||||||
defer reader.Close()
|
defer reader.Close()
|
||||||
if gerr := getObject(ctx, bucket, object, 0, objInfo.Size, writer, objInfo.ETag); gerr != nil {
|
|
||||||
pipewriter.CloseWithError(gerr)
|
startOffset, length := int64(0), objInfo.Size
|
||||||
|
// Get the object.
|
||||||
|
if objectAPI.IsEncryptionSupported() {
|
||||||
|
s3Encrypted := crypto.IsEncrypted(objInfo.UserDefined)
|
||||||
|
if s3Encrypted {
|
||||||
|
var encReader io.Reader
|
||||||
|
encReader, startOffset, length, err = DecryptBlocksRequestR(reader, r, bucket, object, startOffset, length, objInfo, false)
|
||||||
|
if err != nil {
|
||||||
|
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
pipewriter.Close() // Close writer explicitly signaling we wrote all data.
|
// Resulting reader should be limited early on
|
||||||
}()
|
// for decryption upto required length,
|
||||||
|
// additionally also skipping mod(offset)64KiB
|
||||||
|
// boundaries.
|
||||||
|
encReader = io.LimitReader(ioutil.NewSkipReader(encReader, startOffset%(64*1024)), length)
|
||||||
|
cleanUp := func() { reader.Close() }
|
||||||
|
reader = NewGetObjectReader(encReader, nil, cleanUp)
|
||||||
|
if s3Encrypted {
|
||||||
|
w.Header().Set(crypto.SSEHeader, crypto.SSEAlgorithmAES256)
|
||||||
|
} else {
|
||||||
|
w.Header().Set(crypto.SSECAlgorithm, r.Header.Get(crypto.SSECAlgorithm))
|
||||||
|
w.Header().Set(crypto.SSECKeyMD5, r.Header.Get(crypto.SSECKeyMD5))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
//s3select //Options
|
//s3select //Options
|
||||||
if selectReq.OutputSerialization.CSV.FieldDelimiter == "" {
|
if selectReq.OutputSerialization.CSV.FieldDelimiter == "" {
|
||||||
@ -272,6 +299,44 @@ func (api objectAPIHandlers) GetObjectHandler(w http.ResponseWriter, r *http.Req
|
|||||||
bucket := vars["bucket"]
|
bucket := vars["bucket"]
|
||||||
object := vars["object"]
|
object := vars["object"]
|
||||||
|
|
||||||
|
// Check for auth type to return S3 compatible error.
|
||||||
|
// type to return the correct error (NoSuchKey vs AccessDenied)
|
||||||
|
if s3Error := checkRequestAuthType(ctx, r, policy.GetObjectAction, bucket, object); s3Error != ErrNone {
|
||||||
|
if getRequestAuthType(r) == authTypeAnonymous {
|
||||||
|
// As per "Permission" section in
|
||||||
|
// https://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectGET.html
|
||||||
|
// If the object you request does not exist,
|
||||||
|
// the error Amazon S3 returns depends on
|
||||||
|
// whether you also have the s3:ListBucket
|
||||||
|
// permission.
|
||||||
|
// * If you have the s3:ListBucket permission
|
||||||
|
// on the bucket, Amazon S3 will return an
|
||||||
|
// HTTP status code 404 ("no such key")
|
||||||
|
// error.
|
||||||
|
// * if you don’t have the s3:ListBucket
|
||||||
|
// permission, Amazon S3 will return an HTTP
|
||||||
|
// status code 403 ("access denied") error.`
|
||||||
|
if globalPolicySys.IsAllowed(policy.Args{
|
||||||
|
Action: policy.ListBucketAction,
|
||||||
|
BucketName: bucket,
|
||||||
|
ConditionValues: getConditionValues(r, ""),
|
||||||
|
IsOwner: false,
|
||||||
|
}) {
|
||||||
|
getObjectInfo := objectAPI.GetObjectInfo
|
||||||
|
if api.CacheAPI() != nil {
|
||||||
|
getObjectInfo = api.CacheAPI().GetObjectInfo
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err := getObjectInfo(ctx, bucket, object)
|
||||||
|
if toAPIErrorCode(err) == ErrNoSuchKey {
|
||||||
|
s3Error = ErrNoSuchKey
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
writeErrorResponse(w, s3Error, r.URL)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
getObjectNInfo := objectAPI.GetObjectNInfo
|
getObjectNInfo := objectAPI.GetObjectNInfo
|
||||||
if api.CacheAPI() != nil {
|
if api.CacheAPI() != nil {
|
||||||
getObjectNInfo = api.CacheAPI().GetObjectNInfo
|
getObjectNInfo = api.CacheAPI().GetObjectNInfo
|
||||||
@ -297,63 +362,25 @@ func (api objectAPIHandlers) GetObjectHandler(w http.ResponseWriter, r *http.Req
|
|||||||
}
|
}
|
||||||
|
|
||||||
objInfo, reader, err := getObjectNInfo(ctx, bucket, object, rs)
|
objInfo, reader, err := getObjectNInfo(ctx, bucket, object, rs)
|
||||||
if reader != nil {
|
|
||||||
defer reader.Close()
|
|
||||||
}
|
|
||||||
// Before check err value above, we need to check the auth
|
|
||||||
// type to return the correct error (NoSuchKey vs AccessDenied)
|
|
||||||
if s3Error := checkRequestAuthType(ctx, r, policy.GetObjectAction, bucket, object); s3Error != ErrNone {
|
|
||||||
if getRequestAuthType(r) == authTypeAnonymous {
|
|
||||||
// As per "Permission" section in
|
|
||||||
// https://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectGET.html
|
|
||||||
// If the object you request does not exist,
|
|
||||||
// the error Amazon S3 returns depends on
|
|
||||||
// whether you also have the s3:ListBucket
|
|
||||||
// permission.
|
|
||||||
// * If you have the s3:ListBucket permission
|
|
||||||
// on the bucket, Amazon S3 will return an
|
|
||||||
// HTTP status code 404 ("no such key")
|
|
||||||
// error.
|
|
||||||
// * if you don’t have the s3:ListBucket
|
|
||||||
// permission, Amazon S3 will return an HTTP
|
|
||||||
// status code 403 ("access denied") error.`
|
|
||||||
if globalPolicySys.IsAllowed(policy.Args{
|
|
||||||
Action: policy.ListBucketAction,
|
|
||||||
BucketName: bucket,
|
|
||||||
ConditionValues: getConditionValues(r, ""),
|
|
||||||
IsOwner: false,
|
|
||||||
}) {
|
|
||||||
if toAPIErrorCode(err) == ErrNoSuchKey {
|
|
||||||
s3Error = ErrNoSuchKey
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
writeErrorResponse(w, s3Error, r.URL)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
|
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// If object is encrypted, we avoid the cache layer.
|
// If object is encrypted, we avoid the cache layer.
|
||||||
isEncrypted := objectAPI.IsEncryptionSupported() && (crypto.SSEC.IsRequested(r.Header) ||
|
if crypto.IsEncrypted(objInfo.UserDefined) && api.CacheAPI() != nil {
|
||||||
crypto.S3.IsEncrypted(objInfo.UserDefined))
|
|
||||||
if isEncrypted && api.CacheAPI() != nil {
|
|
||||||
// Close the existing reader before re-querying the backend
|
// Close the existing reader before re-querying the backend
|
||||||
if reader != nil {
|
if reader != nil {
|
||||||
reader.Close()
|
reader.Close()
|
||||||
}
|
}
|
||||||
// Query the backend
|
// Query the backend
|
||||||
objInfo, reader, err = objectAPI.GetObjectNInfo(ctx, bucket, object, rs)
|
objInfo, reader, err = objectAPI.GetObjectNInfo(ctx, bucket, object, rs)
|
||||||
if reader != nil {
|
|
||||||
defer reader.Close()
|
|
||||||
}
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
|
writeErrorResponse(w, toAPIErrorCode(err), r.URL)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
defer reader.Close()
|
||||||
|
|
||||||
if objectAPI.IsEncryptionSupported() {
|
if objectAPI.IsEncryptionSupported() {
|
||||||
if _, err = DecryptObjectInfo(&objInfo, r.Header); err != nil {
|
if _, err = DecryptObjectInfo(&objInfo, r.Header); err != nil {
|
||||||
@ -374,8 +401,8 @@ func (api objectAPIHandlers) GetObjectHandler(w http.ResponseWriter, r *http.Req
|
|||||||
|
|
||||||
// Get the object.
|
// Get the object.
|
||||||
if objectAPI.IsEncryptionSupported() {
|
if objectAPI.IsEncryptionSupported() {
|
||||||
s3Encrypted := crypto.S3.IsEncrypted(objInfo.UserDefined)
|
s3Encrypted := crypto.IsEncrypted(objInfo.UserDefined)
|
||||||
if crypto.SSEC.IsRequested(r.Header) || s3Encrypted {
|
if s3Encrypted {
|
||||||
var encReader io.Reader
|
var encReader io.Reader
|
||||||
encReader, startOffset, length, err = DecryptBlocksRequestR(reader, r, bucket, object, startOffset, length, objInfo, false)
|
encReader, startOffset, length, err = DecryptBlocksRequestR(reader, r, bucket, object, startOffset, length, objInfo, false)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
Loading…
x
Reference in New Issue
Block a user