nzinkevich/frostfs-s3-gw
1
0
Fork 0
forked from TrueCloudLab/frostfs-s3-gw
Code Pull requests Activity

Merge pull request #91 from masterSplinter01/feature/47-add-create-bucket

Browse source 
[#47] Add CreateBucket and DeleteBucket
This commit is contained in:
Roman Khimov 2021-06-23 23:46:04 +03:00 committed by GitHub
commit e05d67cedd
9 changed files with 203 additions and 79 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
api/auth/center.go
View file

@ -14,7 +14,7 @@ import (
"github.com/aws/aws-sdk-go/aws/credentials"
v4 "github.com/aws/aws-sdk-go/aws/signer/v4"
"github.com/nspcc-dev/neofs-api-go/pkg/object"
"github.com/nspcc-dev/neofs-api-go/pkg/token"
"github.com/nspcc-dev/neofs-s3-gw/creds/accessbox"
"github.com/nspcc-dev/neofs-s3-gw/creds/tokens"
"github.com/nspcc-dev/neofs-sdk-go/pkg/pool"
"go.uber.org/zap"
@ -25,7 +25,7 @@ var authorizationFieldRegexp = regexp.MustCompile(`AWS4-HMAC-SHA256 Credential=(
type (
// Center is a user authentication interface.
Center interface {
Authenticate(request *http.Request) (*token.BearerToken, error)
Authenticate(request *http.Request) (*accessbox.GateData, error)
}
center struct {
@ -63,7 +63,7 @@ func New(conns pool.Pool, key *ecdsa.PrivateKey) Center {
}
}
func (c *center) Authenticate(r *http.Request) (*token.BearerToken, error) {
func (c *center) Authenticate(r *http.Request) (*accessbox.GateData, error) {
queryValues := r.URL.Query()
if queryValues.Get("X-Amz-Algorithm") == "AWS4-HMAC-SHA256" {
return nil, errors.New("pre-signed form of request is not supported")
@ -127,5 +127,5 @@ func (c *center) Authenticate(r *http.Request) (*token.BearerToken, error) {
return nil, errors.New("failed to pass authentication procedure")
}
return tkns.BearerToken, nil
return tkns, nil
}

23
api/handler/delete.go
View file

@ -6,6 +6,7 @@ import (
"github.com/gorilla/mux"
"github.com/nspcc-dev/neofs-s3-gw/api"
"github.com/nspcc-dev/neofs-s3-gw/api/layer"
"go.uber.org/zap"
"google.golang.org/grpc/status"
)
@ -157,3 +158,25 @@ func (h *handler) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *http.Re
return
}
}
func (h *handler) DeleteBucketHandler(w http.ResponseWriter, r *http.Request) {
var (
rid = api.GetRequestID(r.Context())
p = layer.DeleteBucketParams{}
req = mux.Vars(r)
)
p.Name = req["bucket"]
err := h.obj.DeleteBucket(r.Context(), &p)
if err != nil {
h.log.Error("couldn't delete bucket",
zap.String("request_id", rid),
zap.String("bucket_name", p.Name),
zap.Error(err))
api.WriteErrorResponse(r.Context(), w, api.Error{
Code: api.GetAPIError(api.ErrInternalError).Code,
Description: err.Error(),
HTTPStatusCode: http.StatusInternalServerError,
}, r.URL)
}
}

16
api/handler/not-support.go
View file

@ -7,14 +7,6 @@ import (
"github.com/nspcc-dev/neofs-s3-gw/api"
)
func (h *handler) DeleteBucketHandler(w http.ResponseWriter, r *http.Request) {
api.WriteErrorResponse(r.Context(), w, api.Error{
Code: api.GetAPIError(api.ErrBadRequest).Code,
Description: notSupported + mux.CurrentRoute(r).GetName(),
HTTPStatusCode: http.StatusNotImplemented,
}, r.URL)
}
func (h *handler) DeleteBucketPolicyHandler(w http.ResponseWriter, r *http.Request) {
api.WriteErrorResponse(r.Context(), w, api.Error{
Code: api.GetAPIError(api.ErrBadRequest).Code,
@ -62,11 +54,3 @@ func (h *handler) PutBucketNotificationHandler(w http.ResponseWriter, r *http.Re
HTTPStatusCode: http.StatusNotImplemented,
}, r.URL)
}
func (h *handler) PutBucketHandler(w http.ResponseWriter, r *http.Request) {
api.WriteErrorResponse(r.Context(), w, api.Error{
Code: api.GetAPIError(api.ErrBadRequest).Code,
Description: notSupported + mux.CurrentRoute(r).GetName(),
HTTPStatusCode: http.StatusNotImplemented,
}, r.URL)
}

91
api/handler/put.go
View file

@ -1,14 +1,27 @@
package handler
import (
"fmt"
"net/http"
"strconv"
"strings"
"github.com/gorilla/mux"
"github.com/nspcc-dev/neofs-api-go/pkg/acl"
"github.com/nspcc-dev/neofs-node/pkg/policy"
"github.com/nspcc-dev/neofs-s3-gw/api"
"github.com/nspcc-dev/neofs-s3-gw/api/layer"
"go.uber.org/zap"
)
// keywords of predefined basic ACL values.
const (
basicACLPrivate = "private"
basicACLReadOnly = "public-read"
basicACLPublic = "public-read-write"
defaultPolicy = "REP 3"
)
func (h *handler) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
var (
err error
@ -58,3 +71,81 @@ func (h *handler) PutObjectHandler(w http.ResponseWriter, r *http.Request) {
api.WriteSuccessResponseHeadersOnly(w)
}
func (h *handler) CreateBucketHandler(w http.ResponseWriter, r *http.Request) {
var (
err error
p = layer.CreateBucketParams{}
rid = api.GetRequestID(r.Context())
req = mux.Vars(r)
)
p.Name = req["bucket"]
if val, ok := r.Header["X-Amz-Acl"]; ok {
p.ACL, err = parseBasicACL(val[0])
} else {
p.ACL = acl.PrivateBasicRule
}
if err != nil {
h.log.Error("could not parse basic ACL",
zap.String("request_id", rid),
zap.Error(err))
api.WriteErrorResponse(r.Context(), w, api.Error{
Code: api.GetAPIError(api.ErrBadRequest).Code,
Description: err.Error(),
HTTPStatusCode: http.StatusBadRequest,
}, r.URL)
}
p.Policy, err = policy.Parse(defaultPolicy)
if err != nil {
h.log.Error("could not parse policy",
zap.String("request_id", rid),
zap.Error(err))
api.WriteErrorResponse(r.Context(), w, api.Error{
Code: api.GetAPIError(api.ErrBadRequest).Code,
Description: err.Error(),
HTTPStatusCode: http.StatusBadRequest,
}, r.URL)
}
cid, err := h.obj.CreateBucket(r.Context(), &p)
if err != nil {
h.log.Error("could not create bucket",
zap.String("request_id", rid),
zap.Error(err))
api.WriteErrorResponse(r.Context(), w, api.Error{
Code: api.GetAPIError(api.ErrInternalError).Code,
Description: err.Error(),
HTTPStatusCode: http.StatusInternalServerError,
}, r.URL)
}
h.log.Info("bucket is created",
zap.String("container_id", cid.String()))
api.WriteSuccessResponseHeadersOnly(w)
}
func parseBasicACL(basicACL string) (uint32, error) {
switch basicACL {
case basicACLPublic:
return acl.PublicBasicRule, nil
case basicACLPrivate:
return acl.PrivateBasicRule, nil
case basicACLReadOnly:
return acl.ReadOnlyBasicRule, nil
default:
basicACL = strings.Trim(strings.ToLower(basicACL), "0x")
value, err := strconv.ParseUint(basicACL, 16, 32)
if err != nil {
return 0, fmt.Errorf("can't parse basic ACL: %s", basicACL)
}
return uint32(value), nil
}
}

50
api/layer/container.go
View file

@ -2,6 +2,7 @@ package layer
import (
"context"
"fmt"
"strconv"
"time"
@ -9,6 +10,8 @@ import (
cid "github.com/nspcc-dev/neofs-api-go/pkg/container/id"
"github.com/nspcc-dev/neofs-api-go/pkg/owner"
"github.com/nspcc-dev/neofs-s3-gw/api"
"github.com/nspcc-dev/neofs-s3-gw/creds/accessbox"
"github.com/nspcc-dev/neofs-sdk-go/pkg/pool"
"go.uber.org/zap"
)
@ -43,15 +46,7 @@ func (n *layer) containerInfo(ctx context.Context, cid *cid.ID) (*BucketInfo, er
Name: cid.String(),
}
)
conn, _, err := n.pool.Connection()
if err != nil {
n.log.Error("failed to get connection from the pool",
zap.String("request_id", rid),
zap.Error(err))
return nil, err
}
res, err = conn.GetContainer(ctx, cid, bearerOpt)
res, err = n.pool.GetContainer(ctx, cid, bearerOpt)
if err != nil {
n.log.Error("could not fetch container",
zap.Stringer("cid", cid),
@ -94,15 +89,7 @@ func (n *layer) containerList(ctx context.Context) ([]*BucketInfo, error) {
res []*cid.ID
rid = api.GetRequestID(ctx)
)
conn, _, err := n.pool.Connection()
if err != nil {
n.log.Error("failed to get connection from the pool",
zap.String("request_id", rid),
zap.Error(err))
return nil, err
}
res, err = conn.ListContainers(ctx, own, bearerOpt)
res, err = n.pool.ListContainers(ctx, own, bearerOpt)
if err != nil {
n.log.Error("could not fetch container",
zap.String("request_id", rid),
@ -125,3 +112,30 @@ func (n *layer) containerList(ctx context.Context) ([]*BucketInfo, error) {
return list, nil
}
func (n *layer) createContainer(ctx context.Context, p *CreateBucketParams) (*cid.ID, error) {
cnr := container.New(
container.WithPolicy(p.Policy),
container.WithCustomBasicACL(p.ACL),
container.WithAttribute(container.AttributeName, p.Name),
container.WithAttribute(container.AttributeTimestamp, strconv.FormatInt(time.Now().Unix(), 10)))
cnr.SetSessionToken(ctx.Value(api.GateData).(*accessbox.GateData).SessionToken)
cnr.SetOwnerID(n.Owner(ctx))
cid, err := n.pool.PutContainer(ctx, cnr)
if err != nil {
return nil, fmt.Errorf("failed to create a bucket: %w", err)
}
err = n.pool.WaitForContainerPresence(ctx, cid, pool.DefaultPollingParams())
if err != nil {
return nil, err
}
return cid, nil
}
func (n *layer) deleteContainer(ctx context.Context, cid *cid.ID) error {
return n.pool.DeleteContainer(ctx, cid, n.SessionOpt(ctx))
}

45
api/layer/layer.go
View file

@ -11,10 +11,11 @@ import (
"github.com/nspcc-dev/neofs-api-go/pkg/client"
cid "github.com/nspcc-dev/neofs-api-go/pkg/container/id"
"github.com/nspcc-dev/neofs-api-go/pkg/netmap"
"github.com/nspcc-dev/neofs-api-go/pkg/object"
"github.com/nspcc-dev/neofs-api-go/pkg/owner"
"github.com/nspcc-dev/neofs-api-go/pkg/token"
"github.com/nspcc-dev/neofs-s3-gw/api"
"github.com/nspcc-dev/neofs-s3-gw/creds/accessbox"
"github.com/nspcc-dev/neofs-sdk-go/pkg/pool"
"go.uber.org/zap"
"google.golang.org/grpc/codes"
@ -61,6 +62,16 @@ type (
DstObject string
Header map[string]string
}
// CreateBucketParams stores bucket create request parameters.
CreateBucketParams struct {
Name string
ACL uint32
Policy *netmap.PlacementPolicy
}
// DeleteBucketParams stores delete bucket request parameters.
DeleteBucketParams struct {
Name string
}
// NeoFS provides basic NeoFS interface.
NeoFS interface {
@ -73,6 +84,8 @@ type (
ListBuckets(ctx context.Context) ([]*BucketInfo, error)
GetBucketInfo(ctx context.Context, name string) (*BucketInfo, error)
CreateBucket(ctx context.Context, p *CreateBucketParams) (*cid.ID, error)
DeleteBucket(ctx context.Context, p *DeleteBucketParams) error
GetObject(ctx context.Context, p *GetObjectParams) error
GetObjectInfo(ctx context.Context, bucketName, objectName string) (*ObjectInfo, error)
@ -106,8 +119,8 @@ func NewLayer(log *zap.Logger, conns pool.Pool) Client {
// Owner returns owner id from BearerToken (context) or from client owner.
func (n *layer) Owner(ctx context.Context) *owner.ID {
if tkn, ok := ctx.Value(api.BearerTokenKey).(*token.BearerToken); ok && tkn != nil {
return tkn.Issuer()
if data, ok := ctx.Value(api.GateData).(*accessbox.GateData); ok && data != nil {
return data.BearerToken.Issuer()
}
return n.pool.OwnerID()
@ -115,13 +128,22 @@ func (n *layer) Owner(ctx context.Context) *owner.ID {
// BearerOpt returns client.WithBearer call option with token from context or with nil token.
func (n *layer) BearerOpt(ctx context.Context) client.CallOption {
if tkn, ok := ctx.Value(api.BearerTokenKey).(*token.BearerToken); ok && tkn != nil {
return client.WithBearer(tkn)
if data, ok := ctx.Value(api.GateData).(*accessbox.GateData); ok && data != nil {
return client.WithBearer(data.BearerToken)
}
return client.WithBearer(nil)
}
// SessionOpt returns client.WithSession call option with token from context or with nil token.
func (n *layer) SessionOpt(ctx context.Context) client.CallOption {
if data, ok := ctx.Value(api.GateData).(*accessbox.GateData); ok && data != nil {
return client.WithSession(data.SessionToken)
}
return client.WithSession(nil)
}
// Get NeoFS Object by refs.Address (should be used by auth.Center).
func (n *layer) Get(ctx context.Context, address *object.Address) (*object.Object, error) {
ops := new(client.GetObjectParams).WithAddress(address)
@ -399,3 +421,16 @@ func (n *layer) DeleteObjects(ctx context.Context, bucket string, objects []stri
return errs
}
func (n *layer) CreateBucket(ctx context.Context, p *CreateBucketParams) (*cid.ID, error) {
return n.createContainer(ctx, p)
}
func (n *layer) DeleteBucket(ctx context.Context, p *DeleteBucketParams) error {
bucketInfo, err := n.GetBucketInfo(ctx, p.Name)
if err != nil {
return err
}
return n.deleteContainer(ctx, bucketInfo.CID)
}

8
api/router.go
View file

@ -68,7 +68,7 @@ type (
PutBucketTaggingHandler(http.ResponseWriter, *http.Request)
PutBucketVersioningHandler(http.ResponseWriter, *http.Request)
PutBucketNotificationHandler(http.ResponseWriter, *http.Request)
PutBucketHandler(http.ResponseWriter, *http.Request)
CreateBucketHandler(http.ResponseWriter, *http.Request)
HeadBucketHandler(http.ResponseWriter, *http.Request)
PostPolicyBucketHandler(http.ResponseWriter, *http.Request)
DeleteMultipleObjectsHandler(http.ResponseWriter, *http.Request)
@ -412,10 +412,10 @@ func Attach(r *mux.Router, domains []string, m MaxClients, h Handler, center aut
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbucketnotification", h.PutBucketNotificationHandler))).Queries("notification", "").
Name("PutBucketNotification")
// PutBucket
// CreateBucket
bucket.Methods(http.MethodPut).HandlerFunc(
m.Handle(metrics.APIStats("putbucket", h.PutBucketHandler))).
Name("PutBucket")
m.Handle(metrics.APIStats("createbucket", h.CreateBucketHandler))).
Name("CreateBucket")
// HeadBucket
bucket.Methods(http.MethodHead).HandlerFunc(
m.Handle(metrics.APIStats("headbucket", h.HeadBucketHandler))).

8
api/user-auth.go
View file

@ -12,15 +12,15 @@ import (
// KeyWrapper is wrapper for context keys.
type KeyWrapper string
// BearerTokenKey is an ID used to store bearer token in a context.
var BearerTokenKey = KeyWrapper("__context_bearer_token_key")
// GateData is an ID used to store GateData in a context.
var GateData = KeyWrapper("__context_gate_data_key")
// AttachUserAuth adds user authentication via center to router using log for logging.
func AttachUserAuth(router *mux.Router, center auth.Center, log *zap.Logger) {
router.Use(func(h http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var ctx context.Context
token, err := center.Authenticate(r)
tokens, err := center.Authenticate(r)
if err != nil {
if err == auth.ErrNoAuthorizationHeader {
log.Debug("couldn't receive bearer token, using neofs-key")
@ -31,7 +31,7 @@ func AttachUserAuth(router *mux.Router, center auth.Center, log *zap.Logger) {
return
}
} else {
ctx = context.WithValue(r.Context(), BearerTokenKey, token)
ctx = context.WithValue(r.Context(), GateData, tokens)
}
h.ServeHTTP(w, r.WithContext(ctx))

33
authmate/authmate.go
View file

@ -30,8 +30,6 @@ import (
const (
defaultAuthContainerBasicACL uint32 = 0b00111100100011001000110011001110
containerCreationTimeout = 120 * time.Second
containerPollInterval = 5 * time.Second
)
// Agent contains client communicating with NeoFS and logger.
@ -78,14 +76,9 @@ type (
)
func (a *Agent) checkContainer(ctx context.Context, cid *cid.ID, friendlyName string) (*cid.ID, error) {
conn, _, err := a.pool.Connection()
if err != nil {
return nil, err
}
if cid != nil {
// check that container exists
_, err = conn.GetContainer(ctx, cid)
_, err := a.pool.GetContainer(ctx, cid)
return cid, err
}
@ -100,31 +93,15 @@ func (a *Agent) checkContainer(ctx context.Context, cid *cid.ID, friendlyName st
container.WithAttribute(container.AttributeName, friendlyName),
container.WithAttribute(container.AttributeTimestamp, strconv.FormatInt(time.Now().Unix(), 10)))
cid, err = conn.PutContainer(ctx, cnr)
cid, err = a.pool.PutContainer(ctx, cnr)
if err != nil {
return nil, err
}
wctx, cancel := context.WithTimeout(ctx, containerCreationTimeout)
defer cancel()
ticker := time.NewTimer(containerPollInterval)
defer ticker.Stop()
wdone := wctx.Done()
done := ctx.Done()
for {
select {
case <-done:
return nil, ctx.Err()
case <-wdone:
return nil, wctx.Err()
case <-ticker.C:
_, err = conn.GetContainer(ctx, cid)
if err == nil {
return cid, nil
}
ticker.Reset(containerPollInterval)
}
if err := a.pool.WaitForContainerPresence(ctx, cid, pool.DefaultPollingParams()); err != nil {
return nil, err
}
return cid, nil
}
// IssueSecret creates an auth token, puts it in the NeoFS network and writes to io.Writer a new secret access key.