forked from TrueCloudLab/frostfs-node
[#872] object: Introduce APE middlewar for object service
Signed-off-by: Airat Arifullin <a.arifullin@yadro.com>
This commit is contained in:
parent
e43609c616
commit
c8baf76fae
13 changed files with 1456 additions and 175 deletions
|
|
@ -5,6 +5,7 @@ import (
|
|||
"errors"
|
||||
"fmt"
|
||||
|
||||
containerV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/container"
|
||||
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
|
||||
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/session"
|
||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container"
|
||||
|
|
@ -67,10 +68,6 @@ type cfg struct {
|
|||
|
||||
checker ACLChecker
|
||||
|
||||
// TODO(aarifullin): apeCheck is temporarily the part of
|
||||
// acl service and must be standalone.
|
||||
apeChecker APEChainChecker
|
||||
|
||||
irFetcher InnerRingFetcher
|
||||
|
||||
nm netmap.Source
|
||||
|
|
@ -83,7 +80,6 @@ func New(next object.ServiceServer,
|
|||
nm netmap.Source,
|
||||
irf InnerRingFetcher,
|
||||
acl ACLChecker,
|
||||
apeChecker APEChainChecker,
|
||||
cs container.Source,
|
||||
opts ...Option,
|
||||
) Service {
|
||||
|
|
@ -93,7 +89,6 @@ func New(next object.ServiceServer,
|
|||
nm: nm,
|
||||
irFetcher: irf,
|
||||
checker: acl,
|
||||
apeChecker: apeChecker,
|
||||
containers: cs,
|
||||
}
|
||||
|
||||
|
|
@ -107,6 +102,75 @@ func New(next object.ServiceServer,
|
|||
}
|
||||
}
|
||||
|
||||
// wrappedGetObjectStream propagates RequestContext into GetObjectStream's context.
|
||||
// This allows to retrieve already calculated immutable request-specific values in next handler invocation.
|
||||
type wrappedGetObjectStream struct {
|
||||
object.GetObjectStream
|
||||
|
||||
requestInfo RequestInfo
|
||||
}
|
||||
|
||||
func (w *wrappedGetObjectStream) Context() context.Context {
|
||||
return context.WithValue(w.GetObjectStream.Context(), object.RequestContextKey, &object.RequestContext{
|
||||
Namespace: w.requestInfo.ContainerNamespace(),
|
||||
SenderKey: w.requestInfo.SenderKey(),
|
||||
Role: w.requestInfo.RequestRole(),
|
||||
})
|
||||
}
|
||||
|
||||
func newWrappedGetObjectStreamStream(getObjectStream object.GetObjectStream, reqInfo RequestInfo) object.GetObjectStream {
|
||||
return &wrappedGetObjectStream{
|
||||
GetObjectStream: getObjectStream,
|
||||
requestInfo: reqInfo,
|
||||
}
|
||||
}
|
||||
|
||||
// wrappedRangeStream propagates RequestContext into GetObjectRangeStream's context.
|
||||
// This allows to retrieve already calculated immutable request-specific values in next handler invocation.
|
||||
type wrappedRangeStream struct {
|
||||
object.GetObjectRangeStream
|
||||
|
||||
requestInfo RequestInfo
|
||||
}
|
||||
|
||||
func (w *wrappedRangeStream) Context() context.Context {
|
||||
return context.WithValue(w.GetObjectRangeStream.Context(), object.RequestContextKey, &object.RequestContext{
|
||||
Namespace: w.requestInfo.ContainerNamespace(),
|
||||
SenderKey: w.requestInfo.SenderKey(),
|
||||
Role: w.requestInfo.RequestRole(),
|
||||
})
|
||||
}
|
||||
|
||||
func newWrappedRangeStream(rangeStream object.GetObjectRangeStream, reqInfo RequestInfo) object.GetObjectRangeStream {
|
||||
return &wrappedRangeStream{
|
||||
GetObjectRangeStream: rangeStream,
|
||||
requestInfo: reqInfo,
|
||||
}
|
||||
}
|
||||
|
||||
// wrappedSearchStream propagates RequestContext into SearchStream's context.
|
||||
// This allows to retrieve already calculated immutable request-specific values in next handler invocation.
|
||||
type wrappedSearchStream struct {
|
||||
object.SearchStream
|
||||
|
||||
requestInfo RequestInfo
|
||||
}
|
||||
|
||||
func (w *wrappedSearchStream) Context() context.Context {
|
||||
return context.WithValue(w.SearchStream.Context(), object.RequestContextKey, &object.RequestContext{
|
||||
Namespace: w.requestInfo.ContainerNamespace(),
|
||||
SenderKey: w.requestInfo.SenderKey(),
|
||||
Role: w.requestInfo.RequestRole(),
|
||||
})
|
||||
}
|
||||
|
||||
func newWrappedSearchStream(searchStream object.SearchStream, reqInfo RequestInfo) object.SearchStream {
|
||||
return &wrappedSearchStream{
|
||||
SearchStream: searchStream,
|
||||
requestInfo: reqInfo,
|
||||
}
|
||||
}
|
||||
|
||||
// Get implements ServiceServer interface, makes ACL checks and calls
|
||||
// next Get method in the ServiceServer pipeline.
|
||||
func (b Service) Get(request *objectV2.GetRequest, stream object.GetObjectStream) error {
|
||||
|
|
@ -158,7 +222,7 @@ func (b Service) Get(request *objectV2.GetRequest, stream object.GetObjectStream
|
|||
}
|
||||
|
||||
return b.next.Get(request, &getStreamBasicChecker{
|
||||
GetObjectStream: stream,
|
||||
GetObjectStream: newWrappedGetObjectStreamStream(stream, reqInfo),
|
||||
info: reqInfo,
|
||||
checker: b.checker,
|
||||
})
|
||||
|
|
@ -224,7 +288,7 @@ func (b Service) Head(
|
|||
return nil, eACLErr(reqInfo, err)
|
||||
}
|
||||
|
||||
resp, err := b.next.Head(ctx, request)
|
||||
resp, err := b.next.Head(requestContext(ctx, reqInfo), request)
|
||||
if err == nil {
|
||||
if err = b.checker.CheckEACL(resp, reqInfo); err != nil {
|
||||
err = eACLErr(reqInfo, err)
|
||||
|
|
@ -277,7 +341,7 @@ func (b Service) Search(request *objectV2.SearchRequest, stream object.SearchStr
|
|||
|
||||
return b.next.Search(request, &searchStreamBasicChecker{
|
||||
checker: b.checker,
|
||||
SearchStream: stream,
|
||||
SearchStream: newWrappedSearchStream(stream, reqInfo),
|
||||
info: reqInfo,
|
||||
})
|
||||
}
|
||||
|
|
@ -333,7 +397,7 @@ func (b Service) Delete(
|
|||
return nil, eACLErr(reqInfo, err)
|
||||
}
|
||||
|
||||
return b.next.Delete(ctx, request)
|
||||
return b.next.Delete(requestContext(ctx, reqInfo), request)
|
||||
}
|
||||
|
||||
func (b Service) GetRange(request *objectV2.GetRangeRequest, stream object.GetObjectRangeStream) error {
|
||||
|
|
@ -386,11 +450,18 @@ func (b Service) GetRange(request *objectV2.GetRangeRequest, stream object.GetOb
|
|||
|
||||
return b.next.GetRange(request, &rangeStreamBasicChecker{
|
||||
checker: b.checker,
|
||||
GetObjectRangeStream: stream,
|
||||
GetObjectRangeStream: newWrappedRangeStream(stream, reqInfo),
|
||||
info: reqInfo,
|
||||
})
|
||||
}
|
||||
|
||||
func requestContext(ctx context.Context, reqInfo RequestInfo) context.Context {
|
||||
return context.WithValue(ctx, object.RequestContextKey, &object.RequestContext{
|
||||
SenderKey: reqInfo.SenderKey(),
|
||||
Role: reqInfo.RequestRole(),
|
||||
})
|
||||
}
|
||||
|
||||
func (b Service) GetRangeHash(
|
||||
ctx context.Context,
|
||||
request *objectV2.GetRangeHashRequest,
|
||||
|
|
@ -442,7 +513,7 @@ func (b Service) GetRangeHash(
|
|||
return nil, eACLErr(reqInfo, err)
|
||||
}
|
||||
|
||||
return b.next.GetRangeHash(ctx, request)
|
||||
return b.next.GetRangeHash(requestContext(ctx, reqInfo), request)
|
||||
}
|
||||
|
||||
func (b Service) PutSingle(ctx context.Context, request *objectV2.PutSingleRequest) (*objectV2.PutSingleResponse, error) {
|
||||
|
|
@ -501,7 +572,7 @@ func (b Service) PutSingle(ctx context.Context, request *objectV2.PutSingleReque
|
|||
return nil, eACLErr(reqInfo, err)
|
||||
}
|
||||
|
||||
return b.next.PutSingle(ctx, request)
|
||||
return b.next.PutSingle(requestContext(ctx, reqInfo), request)
|
||||
}
|
||||
|
||||
func (p putStreamBasicChecker) Send(ctx context.Context, request *objectV2.PutRequest) error {
|
||||
|
|
@ -566,9 +637,11 @@ func (p putStreamBasicChecker) Send(ctx context.Context, request *objectV2.PutRe
|
|||
|
||||
reqInfo.obj = obj
|
||||
|
||||
if err := p.source.apeChecker.CheckIfRequestPermitted(reqInfo); err != nil {
|
||||
return err
|
||||
if !p.source.checker.CheckBasicACL(reqInfo) || !p.source.checker.StickyBitCheck(reqInfo, idOwner) {
|
||||
return basicACLErr(reqInfo)
|
||||
}
|
||||
|
||||
ctx = requestContext(ctx, reqInfo)
|
||||
}
|
||||
|
||||
return p.next.Send(ctx, request)
|
||||
|
|
@ -671,6 +744,7 @@ func (b Service) findRequestInfo(req MetaWithToken, idCnr cid.ID, op acl.Op) (in
|
|||
info.operation = op
|
||||
info.cnrOwner = cnr.Value.Owner()
|
||||
info.idCnr = idCnr
|
||||
info.cnrNamespace = cnr.Value.Attribute(containerV2.SysAttributeZone)
|
||||
|
||||
// it is assumed that at the moment the key will be valid,
|
||||
// otherwise the request would not pass validation
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue