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,13 +1,26 @@
+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,
+    form_bearertoken_file,
+    set_eacl,
+    create_eacl, EACLFilters, EACLFilter, EACLHeaderType, create_bearer_token, sign_bearer)
+from helpers.common import ASSETS_DIR, TEST_FILES_DIR
 from helpers.container import (
     REP_2_FOR_3_NODES_PLACEMENT_RULE,
     SINGLE_PLACEMENT_RULE,
-    create_container,
+    create_container, DEFAULT_PLACEMENT_RULE,
 )
-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
@@ -17,6 +30,8 @@
 from neofs_testlib.shell import Shell
 from pytest import FixtureRequest
 
+from neofs_testlib.utils.wallet import get_last_address_from_wallet
+
 
 @pytest.fixture(scope="module")
 @allure.title("Create bearer token for OTHERS with all operations allowed for all containers")
@@ -173,3 +188,107 @@ 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,
+        )
+
+        @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,
+        )