node/acl: fix acl checks for non-container nodes

Split objects are validated and it required to do some additional requests for
it. Such operation can be restricted for a non-container node, so checking for
it is an unsolvable problem, skip ACL checks in such cases.
Closes #2909.

Signed-off-by: Pavel Karpy <[email protected]>

CHANGELOG.md

@@ -13,6 +13,7 @@ Changelog for NeoFS Node
 - Control service's Drop call does not clean metabase (#2822)
 - It was impossible to specify memory amount as "1b" (one byte) in config, default was used instead (#2899)
 - Container session token's lifetime was not checked (#2898)
+- ACL checks for split objects could be forced by a node than might lack access (#2909) 
 
 ### Changed
 - neofs-cli allows several objects deletion at a time (#2774)

cmd/neofs-node/object.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 
 	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/nspcc-dev/neofs-api-go/v2/object"
@@ -312,11 +313,12 @@ func initObjectService(c *cfg) {
 	// every object part in every chain will try to refer to the first part, so caching
 	// should help a lot here
 	const cachedFirstObjectsNumber = 1000
+	objNode := newNodeForObjects(cnrNodes, sPut, c.IsLocalKey)
 
 	aclSvc := v2.New(
 		v2.WithLogger(c.log),
 		v2.WithIRFetcher(newCachedIRFetcher(irFetcher)),
-		v2.WithNetmapSource(c.netMapSource),
+		v2.WithNetmapper(netmapSourceWithNodes{Source: c.netMapSource, n: objNode}),
 		v2.WithContainerSource(
 			c.cfgObject.cnrSource,
 		),
@@ -350,8 +352,6 @@ func initObjectService(c *cfg) {
 		firstSvc = objectService.NewMetricCollector(signSvc, c.metricsCollector)
 	}
 
-	objNode := newNodeForObjects(cnrNodes, sPut, c.IsLocalKey)
-
 	server := objectTransportGRPC.New(firstSvc, objNode)
 
 	for _, srv := range c.cfgGRPC.servers {
@@ -739,3 +739,29 @@ func (c *cfg) IsLocalNodePublicKey(b []byte) bool { return c.IsLocalKey(b) }
 func (c *cfg) GetNodesForObject(addr oid.Address) ([][]netmapsdk.NodeInfo, []uint, error) {
 	return c.cfgObject.containerNodes.getNodesForObject(addr)
 }
+
+type netmapSourceWithNodes struct {
+	netmap.Source
+	n *nodeForObjects
+}
+
+func (n netmapSourceWithNodes) ServerInContainer(cID cid.ID) (bool, error) {
+	var serverInContainer bool
+	err := n.n.ForEachContainerNodePublicKeyInLastTwoEpochs(cID, func(pubKey []byte) bool {
+		if n.n.isLocalPubKey(pubKey) {
+			serverInContainer = true
+			return false
+		}
+		return true
+	})
+	if err != nil {
+		// https://github.com/nspcc-dev/neofs-sdk-go/pull/615
+		if strings.Contains(err.Error(), "not enough nodes to SELECT from") {
+			return false, nil
+		}
+
+		return false, err
+	}
+
+	return serverInContainer, nil
+}

pkg/services/object/acl/v2/opts.go

@@ -2,7 +2,6 @@ package v2
 
 import (
 	"github.com/nspcc-dev/neofs-node/pkg/core/container"
-	"github.com/nspcc-dev/neofs-node/pkg/core/netmap"
 	objectSvc "github.com/nspcc-dev/neofs-node/pkg/services/object"
 	"go.uber.org/zap"
 )
@@ -14,9 +13,9 @@ func WithLogger(v *zap.Logger) Option {
 	}
 }
 
-// WithNetmapSource return option to set
+// WithNetmapper return option to set
 // netmap source.
-func WithNetmapSource(v netmap.Source) Option {
+func WithNetmapper(v Netmapper) Option {
 	return func(c *cfg) {
 		c.nm = v
 	}

pkg/services/object/acl/v2/service.go

@@ -58,6 +58,15 @@ type searchStreamBasicChecker struct {
 // Option represents Service constructor option.
 type Option func(*cfg)
 
+// Netmapper must provide network map information.
+type Netmapper interface {
+	netmap.Source
+	// ServerInContainer checks if current node belongs to requested container.
+	// Any unknown state must be returned as `(false, error.New("explanation"))`,
+	// not `(false, nil)`.
+	ServerInContainer(cid.ID) (bool, error)
+}
+
 type cfg struct {
 	log *zap.Logger
 
@@ -67,7 +76,7 @@ type cfg struct {
 
 	irFetcher InnerRingFetcher
 
-	nm netmap.Source
+	nm Netmapper
 
 	next object.ServiceServer
 }
@@ -458,6 +467,27 @@ func (p putStreamBasicChecker) Send(request *objectV2.PutRequest) error {
 
 		header := part.GetHeader()
 
+		cIDV2 := header.GetContainerID().GetValue()
+		var cID cid.ID
+		err = cID.Decode(cIDV2)
+		if err != nil {
+			return fmt.Errorf("invalid container ID: %w", err)
+		}
+
+		inContainer, err := p.source.nm.ServerInContainer(cID)
+		if err != nil {
+			return fmt.Errorf("checking if node in container: %w", err)
+		}
+
+		if header.GetSplit() != nil && !inContainer {
+			// skip ACL checks for split objects if it is not a container
+			// node, since it requires additional object operations (e.g.,
+			// requesting the other split parts) that may (or may not) be
+			// prohibited by ACL rules; this node is not going to store such
+			// objects anyway
+			return p.next.Send(request)
+		}
+
 		idV2 := header.GetOwnerID()
 		if idV2 == nil {
 			return errors.New("missing object owner")