frostfs-s3-gw/internal/frostfs/authmate.go

package frostfs

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"strconv"
	"strings"
	"time"

	objectv2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer/frostfs"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/middleware"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/authmate"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/creds/tokens"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/frostfs/crdt"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/internal/logs"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/acl"
	cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	"go.uber.org/zap"
)

const (
	accessBoxCRDTNameAttr = "S3-Access-Box-CRDT-Name"
)

// AuthmateFrostFS is a mediator which implements authmate.FrostFS through pool.Pool.
type AuthmateFrostFS struct {
	frostFS frostfs.FrostFS
	log     *zap.Logger
}

// NewAuthmateFrostFS creates new AuthmateFrostFS using provided pool.Pool.
func NewAuthmateFrostFS(frostFS frostfs.FrostFS, log *zap.Logger) *AuthmateFrostFS {
	return &AuthmateFrostFS{frostFS: frostFS, log: log}
}

// ContainerExists implements authmate.FrostFS interface method.
func (x *AuthmateFrostFS) ContainerExists(ctx context.Context, idCnr cid.ID) error {
	_, err := x.frostFS.Container(ctx, frostfs.PrmContainer{ContainerID: idCnr})
	if err != nil {
		return fmt.Errorf("get container via connection pool: %w", err)
	}

	return nil
}

// TimeToEpoch implements authmate.FrostFS interface method.
func (x *AuthmateFrostFS) TimeToEpoch(ctx context.Context, futureTime time.Time) (uint64, uint64, error) {
	return x.frostFS.TimeToEpoch(ctx, time.Now(), futureTime)
}

// CreateContainer implements authmate.FrostFS interface method.
func (x *AuthmateFrostFS) CreateContainer(ctx context.Context, prm authmate.PrmContainerCreate) (cid.ID, error) {
	basicACL := acl.Private
	// allow reading objects to OTHERS in order to provide read access to S3 gateways
	basicACL.AllowOp(acl.OpObjectGet, acl.RoleOthers)
	basicACL.AllowOp(acl.OpObjectHead, acl.RoleOthers)
	basicACL.AllowOp(acl.OpObjectSearch, acl.RoleOthers)

	res, err := x.frostFS.CreateContainer(ctx, frostfs.PrmContainerCreate{
		Creator:  prm.Owner,
		Policy:   prm.Policy,
		Name:     prm.FriendlyName,
		BasicACL: basicACL,
	})
	if err != nil {
		return cid.ID{}, err
	}
	return res.ContainerID, nil
}

// GetCredsObject implements authmate.FrostFS interface method.
func (x *AuthmateFrostFS) GetCredsObject(ctx context.Context, prm tokens.PrmGetCredsObject) (obj *object.Object, err error) {
	var readObjAddr *oid.Address
	defer func() {
		if prm.FallbackAddress == nil {
			return
		}
		if err != nil && (readObjAddr == nil || !readObjAddr.Equals(*prm.FallbackAddress)) {
			obj, err = x.readObject(ctx, *prm.FallbackAddress)
		}
	}()

	versions, err := x.getCredVersions(ctx, prm.Container, prm.AccessKeyID)
	if err != nil {
		return nil, err
	}

	var addr oid.Address
	isCustom := addr.DecodeString(strings.ReplaceAll(prm.AccessKeyID, "0", "/")) != nil

	var credObjID oid.ID
	if last := versions.GetLast(); last != nil {
		credObjID = last.ObjID
	} else if !isCustom {
		credObjID = addr.Object()
	} else {
		return nil, fmt.Errorf("%w: '%s'", tokens.ErrCustomAccessKeyIDNotFound, prm.AccessKeyID)
	}

	readObjAddr = &oid.Address{}
	readObjAddr.SetContainer(prm.Container)
	readObjAddr.SetObject(credObjID)

	return x.readObject(ctx, *readObjAddr)
}

func (x *AuthmateFrostFS) readObject(ctx context.Context, addr oid.Address) (*object.Object, error) {
	res, err := x.frostFS.GetObject(ctx, frostfs.PrmObjectGet{
		Container: addr.Container(),
		Object:    addr.Object(),
	})
	if err != nil {
		return nil, err
	}

	defer func() {
		if closeErr := res.Payload.Close(); closeErr != nil {
			x.reqLogger(ctx).Warn(logs.CloseCredsObjectPayload, zap.Error(closeErr))
		}
	}()

	data, err := io.ReadAll(res.Payload)
	if err != nil {
		return nil, err
	}
	res.Header.SetPayload(data)

	return &res.Header, err
}

// CreateObject implements authmate.FrostFS interface method.
func (x *AuthmateFrostFS) CreateObject(ctx context.Context, prm tokens.PrmObjectCreate) (oid.ID, error) {
	attributes := [][2]string{{objectv2.SysAttributeExpEpoch, strconv.FormatUint(prm.ExpirationEpoch, 10)}}

	if prm.NewVersionForAccessKeyID != "" {
		versions, err := x.getCredVersions(ctx, prm.Container, prm.NewVersionForAccessKeyID)
		if err != nil {
			return oid.ID{}, err
		}

		if versions.GetLast() == nil {
			var addr oid.Address
			isCustom := addr.DecodeString(strings.ReplaceAll(prm.NewVersionForAccessKeyID, "0", "/")) != nil

			if isCustom {
				return oid.ID{}, fmt.Errorf("creds object for accessKeyId '%s' not found", prm.NewVersionForAccessKeyID)
			}

			versions.AppendVersion(&crdt.ObjectVersion{ObjID: addr.Object()})
		}

		for key, val := range versions.GetCRDTHeaders() {
			attributes = append(attributes, [2]string{key, val})
		}

		attributes = append(attributes, [2]string{accessBoxCRDTNameAttr, versions.Name()})
	} else if prm.CustomAccessKey != "" {
		attributes = append(attributes, [2]string{accessBoxCRDTNameAttr, prm.CustomAccessKey})
	}

	for _, attr := range prm.CustomAttributes {
		// we don't check attribute duplication since storage node does this
		attributes = append(attributes, [2]string{attr.Key(), attr.Value()})
	}

	res, err := x.frostFS.CreateObject(ctx, frostfs.PrmObjectCreate{
		Container:  prm.Container,
		Filepath:   prm.Filepath,
		Attributes: attributes,
		Payload:    bytes.NewReader(prm.Payload),
	})
	if err != nil {
		return oid.ID{}, err
	}

	return res.ObjectID, nil
}

func (x *AuthmateFrostFS) getCredVersions(ctx context.Context, cnrID cid.ID, accessKeyID string) (*crdt.ObjectVersions, error) {
	credVersions, err := x.frostFS.SearchObjects(ctx, frostfs.PrmObjectSearch{
		Container:      cnrID,
		ExactAttribute: [2]string{accessBoxCRDTNameAttr, accessKeyID},
	})
	if err != nil {
		return nil, fmt.Errorf("search s3 access boxes: %w", err)
	}

	versions := crdt.NewObjectVersions(accessKeyID)

	for _, id := range credVersions {
		objVersion, err := x.frostFS.HeadObject(ctx, frostfs.PrmObjectHead{
			Container: cnrID,
			Object:    id,
		})
		if err != nil {
			return nil, fmt.Errorf("head crdt access box '%s': %w", id.EncodeToString(), err)
		}

		versions.AppendVersion(crdt.NewObjectVersion(objVersion))
	}

	return versions, nil
}

func (x *AuthmateFrostFS) reqLogger(ctx context.Context) *zap.Logger {
	reqLogger := middleware.GetReqLog(ctx)
	if reqLogger != nil {
		return reqLogger
	}
	return x.log
}