frostfs-s3-gw/api/handler/api.go

package handler

import (
	"encoding/xml"
	"errors"
	"fmt"
	"io"
	"strconv"
	"strings"
	"time"

	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/logs"
	cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
	"git.frostfs.info/TrueCloudLab/policy-engine/pkg/chain"
	"git.frostfs.info/TrueCloudLab/policy-engine/pkg/engine"
	"go.uber.org/zap"
)

type (
	handler struct {
		log         *zap.Logger
		obj         layer.Client
		notificator Notificator
		cfg         Config
		ape         APE
		frostfsid   FrostFSID
	}

	Notificator interface {
		SendNotifications(topics map[string]string, p *SendNotificationParams) error
		SendTestNotification(topic, bucketName, requestID, HostID string, now time.Time) error
	}

	// Config contains data which handler needs to keep.
	Config interface {
		DefaultPlacementPolicy(namespace string) netmap.PlacementPolicy
		PlacementPolicy(namespace, constraint string) (netmap.PlacementPolicy, bool)
		CopiesNumbers(namespace, constraint string) ([]uint32, bool)
		DefaultCopiesNumbers(namespace string) []uint32
		NewXMLDecoder(io.Reader) *xml.Decoder
		DefaultMaxAge() int
		NotificatorEnabled() bool
		ResolveZoneList() []string
		IsResolveListAllow() bool
		BypassContentEncodingInChunks() bool
		MD5Enabled() bool
	}

	FrostFSID interface {
		GetUserAddress(account, user string) (string, error)
	}

	// APE is Access Policy Engine that needs to save policy and acl info to different places.
	APE interface {
		MorphRuleChainStorage
		PolicyStorage
	}

	// MorphRuleChainStorage is a similar to engine.MorphRuleChainStorage
	// but doesn't know anything about tx.
	MorphRuleChainStorage interface {
		AddChain(target engine.Target, c *chain.Chain) error
		RemoveChain(target engine.Target, chainID chain.ID) error
		ListChains(target engine.Target) ([]*chain.Chain, error)
	}

	// PolicyStorage is interface to save intact initial user provided policy.
	PolicyStorage interface {
		PutPolicy(namespace string, cnrID cid.ID, policy []byte) error
		GetPolicy(namespace string, cnrID cid.ID) ([]byte, error)
		DeletePolicy(namespace string, cnrID cid.ID) error
	}

	frostfsIDDisabled struct{}
)

var _ api.Handler = (*handler)(nil)

// New creates new api.Handler using given logger and client.
func New(log *zap.Logger, obj layer.Client, notificator Notificator, cfg Config, storage APE, ffsid FrostFSID) (api.Handler, error) {
	switch {
	case obj == nil:
		return nil, errors.New("empty FrostFS Object Layer")
	case log == nil:
		return nil, errors.New("empty logger")
	case storage == nil:
		return nil, errors.New("empty policy storage")
	}

	if ffsid == nil {
		ffsid = frostfsIDDisabled{}
	}

	if !cfg.NotificatorEnabled() {
		log.Warn(logs.NotificatorIsDisabledS3WontProduceNotificationEvents)
	} else if notificator == nil {
		return nil, errors.New("empty notificator")
	}

	return &handler{
		log:         log,
		obj:         obj,
		cfg:         cfg,
		ape:         storage,
		notificator: notificator,
		frostfsid:   ffsid,
	}, nil
}

func (f frostfsIDDisabled) GetUserAddress(_, _ string) (string, error) {
	return "", errors.New("frostfsid disabled")
}

// pickCopiesNumbers chooses the return values following this logic:
// 1) array of copies numbers sent in request's header has the highest priority.
// 2) array of copies numbers with corresponding location constraint provided in the config file.
// 3) default copies number from the config file wrapped into array.
func (h *handler) pickCopiesNumbers(metadata map[string]string, namespace, locationConstraint string) ([]uint32, error) {
	copiesNumbersStr, ok := metadata[layer.AttributeFrostfsCopiesNumber]
	if ok {
		result, err := parseCopiesNumbers(copiesNumbersStr)
		if err != nil {
			return nil, err
		}
		return result, nil
	}

	copiesNumbers, ok := h.cfg.CopiesNumbers(namespace, locationConstraint)
	if ok {
		return copiesNumbers, nil
	}

	return h.cfg.DefaultCopiesNumbers(namespace), nil
}

func parseCopiesNumbers(copiesNumbersStr string) ([]uint32, error) {
	var result []uint32
	copiesNumbersSplit := strings.Split(copiesNumbersStr, ",")

	for i := range copiesNumbersSplit {
		item := strings.ReplaceAll(copiesNumbersSplit[i], " ", "")
		if len(item) == 0 {
			continue
		}
		copiesNumber, err := strconv.ParseUint(item, 10, 32)
		if err != nil {
			return nil, fmt.Errorf("pasrse copies number: %w", err)
		}
		result = append(result, uint32(copiesNumber))
	}

	return result, nil
}