tests/object: Add attribute-based ACL test

This is an important case since this is the key functionality for
https://send.fs.neo.org. Closes #786.

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

pytest_tests/tests/object/test_object_api_bearer.py

@@ -1,20 +1,41 @@
+import os
+import uuid
+
 import allure
 import neofs_env.neofs_epoch as neofs_epoch
 import pytest
-from helpers.acl import EACLAccess, EACLOperation, EACLRole, EACLRule, form_bearertoken_file
+from helpers.acl import (
+    EACLAccess,
+    EACLOperation,
+    EACLRole,
+    EACLRule,
+    EACLFilters,
+    EACLFilter,
+    EACLHeaderType,
+    form_bearertoken_file,
+    set_eacl,
+    create_eacl,
+    create_bearer_token,
+    sign_bearer,
+)
+from helpers.common import ASSETS_DIR, TEST_FILES_DIR
 from helpers.container import (
+    DEFAULT_PLACEMENT_RULE,
     REP_2_FOR_3_NODES_PLACEMENT_RULE,
     SINGLE_PLACEMENT_RULE,
     create_container,
 )
-from helpers.neofs_verbs import delete_object, get_object
+from helpers.file_helper import generate_file
+from helpers.grpc_responses import OBJECT_ACCESS_DENIED
+from helpers.neofs_verbs import delete_object, get_object, put_object
 from helpers.storage_container import StorageContainer, StorageContainerInfo
 from helpers.storage_object_info import StorageObjectInfo
 from helpers.test_control import expect_not_raises
 from helpers.wellknown_acl import EACL_PUBLIC_READ_WRITE
 from neofs_env.neofs_env_test_base import NeofsEnvTestBase
 from neofs_testlib.env.env import NeoFSEnv, NodeWallet
 from neofs_testlib.shell import Shell
+from neofs_testlib.utils.wallet import get_last_address_from_wallet
 from pytest import FixtureRequest
 
 
@@ -168,3 +189,109 @@ def test_get_object_with_s3_wallet_bearer_from_all_nodes(
                         bearer=bearer_token_file_all_allow,
                         wallet_config=self.neofs_env.generate_cli_config(s3_gate_wallet),
                     )
+
+    @pytest.mark.parametrize(
+        "user_container",
+        [DEFAULT_PLACEMENT_RULE],
+        indirect=True,
+    )
+    @pytest.mark.parametrize(
+        "file_size",
+        [pytest.lazy_fixture("simple_object_size"), pytest.lazy_fixture("complex_object_size")],
+        ids=["simple object", "complex object"],
+    )
+    def test_attributes_bearer_rules(
+        self,
+        default_wallet: NodeWallet,
+        file_size: int,
+        user_container: StorageContainer,
+    ):
+        # what? user_container has "s3 GW wallet to test bearer", so much magic...
+        other_wallet = user_container.get_wallet_path()
+        container_owner = default_wallet.path
+        cid = user_container.get_id()
+        test_file = generate_file(file_size)
+        ATTRIBUTE_KEY = "test_attribute"
+        ATTRIBUTE_VALUE = "allowed_value"
+
+        with allure.step("Create eACL that prohibits PUT operation for OTHERS"):
+            eacl_deny = create_eacl(
+                cid,
+                [
+                    EACLRule(
+                        role=EACLRole.OTHERS,
+                        access=EACLAccess.DENY,
+                        operation=EACLOperation.PUT,
+                    )
+                ],
+                shell=self.shell,
+            )
+
+        with allure.step("Set container-wise eACL"):
+            set_eacl(
+                container_owner,
+                cid,
+                eacl_deny,
+                shell=self.shell,
+                endpoint=self.neofs_env.sn_rpc,
+            )
+
+        with allure.step("Try to PUT object with OTHER role"):
+            with pytest.raises(Exception, match=OBJECT_ACCESS_DENIED):
+                put_object(
+                    wallet=other_wallet,
+                    path=test_file,
+                    cid=cid,
+                    shell=self.shell,
+                    endpoint=user_container.neofs_env.sn_rpc,
+                )
+
+        with allure.step(f"Create exception for {{{ATTRIBUTE_KEY}: {ATTRIBUTE_VALUE}}} objects"):
+            eacl_allow_exception = create_eacl(
+                cid,
+                [
+                    EACLRule(
+                        role=EACLRole.OTHERS,
+                        access=EACLAccess.ALLOW,
+                        operation=EACLOperation.PUT,
+                        filters=EACLFilters(
+                            filters=[
+                                EACLFilter(key=ATTRIBUTE_KEY, value=ATTRIBUTE_VALUE, header_type=EACLHeaderType.OBJECT)
+                            ]
+                        ),
+                    )
+                ],
+                shell=self.shell,
+            )
+
+        path_to_bearer = os.path.join(os.getcwd(), ASSETS_DIR, TEST_FILES_DIR, f"bearer_token_{str(uuid.uuid4())}")
+
+        create_bearer_token(
+            self.shell,
+            issued_at=1,
+            not_valid_before=1,
+            owner=get_last_address_from_wallet(other_wallet, user_container.neofs_env.default_password),
+            out=path_to_bearer,
+            rpc_endpoint=self.neofs_env.sn_rpc,
+            eacl=eacl_allow_exception,
+            expire_at=(1 << 32) - 1,
+        )
+
+        sign_bearer(
+            shell=self.shell,
+            wallet_path=container_owner,
+            eacl_rules_file_from=path_to_bearer,
+            eacl_rules_file_to=path_to_bearer,
+            json=True,
+        )
+
+        with allure.step("Try to PUT object with exceptional bearer"):
+            put_object(
+                wallet=other_wallet,
+                path=test_file,
+                cid=cid,
+                shell=self.shell,
+                attributes={ATTRIBUTE_KEY: ATTRIBUTE_VALUE},
+                bearer=path_to_bearer,
+                endpoint=user_container.neofs_env.sn_rpc,
+            )