From 085afe2f03e4de9a00c57be059aa6de96d29e65b Mon Sep 17 00:00:00 2001
From: Ekaterina Chernitsyna <e.chernitsyna@yadro.com>
Date: Thu, 10 Oct 2024 15:18:57 +0300
Subject: [PATCH] [#303] add morph and local deny ape tests
---
pytest_tests/testsuites/ape/test_ape.py | 1282 +++++++++++++++++++++++
1 file changed, 1282 insertions(+)
create mode 100644 pytest_tests/testsuites/ape/test_ape.py
diff --git a/pytest_tests/testsuites/ape/test_ape.py b/pytest_tests/testsuites/ape/test_ape.py
new file mode 100644
index 00000000..fc04acf1
--- /dev/null
+++ b/pytest_tests/testsuites/ape/test_ape.py
@@ -0,0 +1,1282 @@
+import allure
+import pytest
+from frostfs_testlib.cli import FrostfsAdm, FrostfsCli
+from frostfs_testlib.credentials.interfaces import User
+from frostfs_testlib.reporter import get_reporter
+from frostfs_testlib.resources.cli import FROSTFS_ADM_CONFIG_PATH, FROSTFS_ADM_EXEC, FROSTFS_CLI_EXEC
+from frostfs_testlib.resources.error_patterns import (
+ NO_RULE_FOUND_OBJECT,
+ OBJECT_ACCESS_DENIED,
+ RULE_ACCESS_DENIED_CONTAINER,
+ RULE_ACCESS_DENIED_OBJECT,
+)
+from frostfs_testlib.shell.interfaces import Shell
+from frostfs_testlib.steps.cli.object import delete_object, get_object, get_range, get_range_hash, head_object, put_object, search_object
+from frostfs_testlib.storage.cluster import Cluster, ClusterNode
+from frostfs_testlib.storage.dataclasses.ape import Operations
+from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
+from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
+from frostfs_testlib.testing.parallel import parallel
+from frostfs_testlib.testing.test_control import expect_not_raises
+from frostfs_testlib.utils.file_utils import generate_file
+
+reporter = get_reporter()
+
+
+@pytest.fixture(scope="session")
+def remote_frostfs_cli_first_node(cluster: Cluster):
+ node = cluster.cluster_nodes[0]
+ shell = node.host.get_shell()
+ cli = FrostfsCli(
+ shell=shell,
+ frostfs_cli_exec_path=FROSTFS_CLI_EXEC,
+ config_file=node.storage_node.get_remote_wallet_config_path(),
+ )
+ return cli
+
+
+@pytest.fixture(scope="session")
+def remote_frostfs_adm_first_node(cluster: Cluster):
+ node = cluster.cluster_nodes[0]
+ shell = node.host.get_shell()
+ adm = FrostfsAdm(
+ shell=shell,
+ frostfs_adm_exec_path=FROSTFS_ADM_EXEC,
+ config_file=FROSTFS_ADM_CONFIG_PATH,
+ )
+ return adm
+
+
+def local_overrides_on_node(node: ClusterNode):
+ target = "Chain ID"
+ shell: Shell = node.host.get_shell()
+ remote_config: str = node.storage_node.get_remote_wallet_config_path()
+ cli = FrostfsCli(shell=shell, frostfs_cli_exec_path=FROSTFS_CLI_EXEC, config_file=remote_config)
+ with reporter.step(f"Check local overrides on {node.storage_node.id} node"):
+ rules = cli.control.list_rules(
+ endpoint=node.storage_node.get_control_endpoint(), target_name="root", target_type="namespace"
+ ).stdout
+ if target in rules:
+ with reporter.step("Delete rules"):
+ chain_ids = [i.split(" ")[2].strip() for i in rules.split("\n") if "Chain ID" in i]
+ for chain_id in chain_ids:
+ cli.control.remove_rule(
+ endpoint=node.storage_node.get_control_endpoint(),
+ target_type="namespace",
+ target_name="root",
+ chain_id=chain_id,
+ )
+
+
+@pytest.fixture(scope="session")
+def remove_rule_ape_in_system(cluster: Cluster) -> None:
+ yield
+ with reporter.step("Check local overrides on nodes."):
+ parallel(local_overrides_on_node, cluster.cluster_nodes)
+
+
+@pytest.mark.ape
+@pytest.mark.ape_morph_rule
+class TestApeMorphRuleChain(ClusterTestBase):
+ @allure.title("MorphRuleChain: Deny to GetContainer in root tenant")
+ def test_morph_rule_chain_deny_to_get_container_root(
+ self,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ ):
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a namespace rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerGet",
+ rule="deny Container.Get *",
+ )
+
+ with reporter.step("Check get the container property on the first node, expected access drnied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.GET_CONTAINER)):
+ frostfs_cli.container.get(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid)
+
+ with reporter.step("Check et the container property on the second node, expected access drnied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.GET_CONTAINER)):
+ frostfs_cli.container.get(self.cluster.storage_nodes[1].get_rpc_endpoint(), cid)
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerGet",
+ )
+
+ with reporter.step("Check get the container property on the first node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.get(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid)
+
+ @allure.title("MorphRuleChain: Deny to PutContainer in root tenant")
+ def test_morph_rule_chain_deny_to_put_container_root(
+ self,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ ):
+
+ with reporter.step("Create a namespace rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerPut",
+ rule="deny Container.Put *",
+ )
+
+ with reporter.step("Check create a container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.PUT_CONTAINER)):
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+
+ with reporter.step("Check create a container on the second node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.PUT_CONTAINER)):
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[1].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl2",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerPut",
+ )
+
+ with reporter.step("Check create a container on the first node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl3",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+
+ @allure.title("Deny to ListContainer in root tenant")
+ def test_morph_rule_chain_deny_to_list_container_root(
+ self,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ ):
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a namespace rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerList",
+ rule="deny Container.List *",
+ )
+
+ with reporter.step("Check list the container properties on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.LIST_CONTAINER)):
+ frostfs_cli.container.list(rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), ttl=1)
+
+ with reporter.step("Check list the container properties on the second node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.LIST_CONTAINER)):
+ frostfs_cli.container.list(rpc_endpoint=self.cluster.storage_nodes[1].get_rpc_endpoint(), ttl=1)
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerList",
+ )
+
+ with reporter.step("Check display a list of containers on the first node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.list(rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), ttl=1)
+
+ @allure.title("MorphRuleChain: Deny to DeleteContainer in root tenant")
+ def test_morph_rule_chain_deny_to_delete_container_root(
+ self,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ ):
+ with reporter.step("Create a namespace rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerDelete",
+ rule="deny Container.Delete *",
+ )
+
+ with reporter.step("Create container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Check delete first container from the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.DELETE_CONTAINER)):
+ frostfs_cli.container.delete(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid, ttl=1)
+
+ with reporter.step("Check delete a second container from the second node, expected access error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.DELETE_CONTAINER)):
+ frostfs_cli.container.delete(self.cluster.storage_nodes[1].get_rpc_endpoint(), cid, ttl=1)
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerDelete",
+ )
+
+ with reporter.step("Check delete first container from the first node,expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.delete(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid, ttl=1)
+
+ @allure.title("MorphRuleChain: Deny to GetObject in root tenant")
+ def test_morph_rule_chain_deny_to_get_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyGetObject",
+ rule=f"deny Object.Get /{cid}/*",
+ )
+
+ with reporter.step("Put object in container on the first node"):
+ oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check get object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.GET_OBJECT)):
+ get_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check get object from container on the second node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.GET_OBJECT)):
+ get_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyGetObject",
+ )
+
+ with reporter.step("Check get object in container on the first node, expected allow"):
+ with expect_not_raises():
+ get_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("MorphRuleChain: Deny to PutObject in root tenant")
+ def test_morph_rule_chain_deny_to_put_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 2 CBF 1",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyPutObject",
+ rule=f"deny Object.Put /{cid}/*",
+ )
+
+ with reporter.step("Check put object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=OBJECT_ACCESS_DENIED):
+ put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check put object from container on the second node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=OBJECT_ACCESS_DENIED):
+ put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyPutObject",
+ )
+
+ with reporter.step("Check put object in container on the first node, expected allow"):
+ with expect_not_raises():
+ put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("MorphRuleChain: Deny to HeadObject in root tenant")
+ def test_morph_rule_chain_deny_to_head_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyHeadObject",
+ rule=f"deny Object.Head /{cid}/*",
+ )
+
+ with reporter.step("Put object in container on the first node"):
+ oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check head object from container on the first nodel expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HEAD_OBJECT)):
+ head_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check head object from container on the second node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HEAD_OBJECT)):
+ head_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyHeadObject",
+ )
+
+ with reporter.step("Check head object in container on the first node, expected allow"):
+ with expect_not_raises():
+ head_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("MorphRuleChain: Deny to SearchObject in root tenant")
+ def test_morph_rule_chain_deny_to_search_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 2",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denySearchObject",
+ rule=f"deny Object.Search /{cid}/*",
+ )
+
+ with reporter.step("Check search object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.SEARCH_OBJECT)):
+ search_object(default_user.wallet, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check search object from container on the second node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.SEARCH_OBJECT)):
+ search_object(default_user.wallet, cid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denySearchObject",
+ )
+
+ with reporter.step("Check search object in container on the first node, expected allow"):
+ with expect_not_raises():
+ search_object(default_user.wallet, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("MorphRuleChain: Deny to RangeObject in root tenant")
+ def test_morph_rule_chain_deny_to_range_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 2",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyRangeObject",
+ rule=f"deny Object.Range /{cid}/*",
+ )
+
+ with reporter.step("Put object in container on the first node"):
+ oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check range object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.RANGE_OBJECT)):
+ get_range(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check range object from container on the second node, expected denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.RANGE_OBJECT)):
+ get_range(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyRangeObject",
+ )
+
+ with reporter.step("Check range object in container on the first node,expected allow"):
+ with expect_not_raises():
+ get_range(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("MorphRuleChain: Deny to HashObject in root tenant")
+ def test_morph_rule_chain_deny_to_hash_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 2",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyHashObject",
+ rule=f"deny Object.Hash /{cid}/*",
+ )
+
+ with reporter.step("Put object in container on the first node"):
+ oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check range hash object from container on the first node, expected denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HASH_OBJECT)):
+ get_range_hash(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check range hash object from container on the second node, expected denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HASH_OBJECT)):
+ get_range_hash(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyHashObject",
+ )
+
+ with reporter.step("Check range hash object in container on the first node, expected denied error"):
+ with expect_not_raises():
+ get_range_hash(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("MorphRuleChain: Deny to DeleteObject in root tenant")
+ def test_morph_rule_chain_deny_to_delete_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_adm_first_node: FrostfsAdm,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 2",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_adm_first_node.morph.add_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyDeleteObject",
+ rule=f"deny Object.Delete /{cid}/*",
+ )
+
+ with reporter.step("Put two objects in container on the first node"):
+ oid_1 = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ oid_2 = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Search object in container on the first node"):
+ search_object_in_container_1 = search_object(
+ default_user.wallet, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()
+ )
+ assert oid_1 in search_object_in_container_1, f"Object {oid_1} was not found"
+ assert oid_2 in search_object_in_container_1, f"Object {oid_2} was not found"
+
+ with reporter.step("Search object from container on the second node"):
+ search_object_in_container_2 = search_object(
+ default_user.wallet, cid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint()
+ )
+ assert oid_1 in search_object_in_container_2, f"Object {oid_1} was not found"
+ assert oid_2 in search_object_in_container_2, f"Object {oid_2} was not found"
+
+ with reporter.step("Check delete object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.DELETE_OBJECT)):
+ delete_object(default_user.wallet, cid, oid_1, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Delete object from container on the second node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.DELETE_OBJECT)):
+ delete_object(default_user.wallet, cid, oid_2, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_adm_first_node.morph.remove_rule(
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyDeleteObject",
+ )
+
+ with reporter.step("Check delete object in container on the first node, expected allow"):
+ with expect_not_raises():
+ delete_object(default_user.wallet, cid, oid_1, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+
+@pytest.mark.ape
+@pytest.mark.ape_local
+class TestApeLocalOverride(ClusterTestBase):
+ @allure.title("LocalOverride: Deny to GetContainer in root tenant")
+ def test_local_override_deny_to_get_container_root(
+ self,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ remove_rule_ape_in_system: None,
+ ):
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a namespace rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerGet",
+ rule="deny Container.Get *",
+ )
+
+ with reporter.step("Check get the container property on the first node, expected denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.GET_CONTAINER)):
+ frostfs_cli.container.get(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid)
+
+ with reporter.step("Check get the container property on the second node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.get(self.cluster.storage_nodes[1].get_rpc_endpoint(), cid)
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerGet",
+ )
+
+ with reporter.step("Check get the container property on the first node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.get(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid)
+
+ @allure.title("LocalOverride: Deny to PutContainer in root tenant")
+ def test_local_override_deny_to_put_container_root(
+ self,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ remove_rule_ape_in_system: None,
+ ):
+
+ with reporter.step("Create a namespace rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerPut",
+ rule="deny Container.Put *",
+ )
+
+ with reporter.step("Check create container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.PUT_CONTAINER)):
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+
+ with reporter.step("Check create a container on the second node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[1].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl2",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerPut",
+ )
+
+ with reporter.step("Check create a container on the first node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl3",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+
+ @allure.title("LocalOverride: Deny to ListContainer in root tenant")
+ def test_local_override_deny_to_list_container_root(
+ self,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ remove_rule_ape_in_system: None,
+ ):
+ with reporter.step("Create a namespace rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerList",
+ rule="deny Container.List *",
+ )
+
+ with reporter.step("Check list the container properties on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.LIST_CONTAINER)):
+ frostfs_cli.container.list(rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), ttl=1)
+
+ with reporter.step("Check list the container properties on the second node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.list(rpc_endpoint=self.cluster.storage_nodes[1].get_rpc_endpoint(), ttl=1)
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerList",
+ )
+
+ with reporter.step("Check display a list of containers on the first node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.list(rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(), ttl=1)
+
+ @allure.title("LocalOverride: Deny to DeleteContainer in root tenant")
+ def test_local_override_deny_to_delete_container_root(
+ self,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ remove_rule_ape_in_system: None,
+ ):
+
+ with reporter.step("Create a namespace rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerDelete",
+ rule="deny Container.Delete *",
+ )
+
+ with reporter.step("Create containers on the first node"):
+ cid_1 = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ cid_2 = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl2",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Check delete first container from the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_CONTAINER.format(operation=Operations.DELETE_CONTAINER)):
+ frostfs_cli.container.delete(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid_1, ttl=1)
+
+ with reporter.step("Check delete a second container from the second node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.delete(self.cluster.storage_nodes[1].get_rpc_endpoint(), cid_2, ttl=1)
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="namespace",
+ target_name="root",
+ chain_id="denyContainerDelete",
+ )
+
+ with reporter.step("Check delete first container from the first node, expected allow"):
+ with expect_not_raises():
+ frostfs_cli.container.delete(self.cluster.storage_nodes[0].get_rpc_endpoint(), cid_1, ttl=1)
+
+ @allure.title("LocalOverride: Deny to GetObject in root tenant")
+ def test_local_override_deny_to_get_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyGetObject",
+ rule=f"deny Object.Get /{cid}/*",
+ )
+
+ with reporter.step("Put object in container on the first node"):
+ oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check get object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.GET_OBJECT)):
+ get_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check get object from container on the second node, expected allow"):
+ with expect_not_raises():
+ get_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyGetObject",
+ )
+
+ with reporter.step("Check get object in container on the first node, expected allow"):
+ with expect_not_raises():
+ get_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("LocalOverride: Deny to PutObject in root tenant")
+ def test_local_override_deny_to_put_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 4",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyPutObject",
+ rule=f"deny Object.Put /{cid}/*",
+ )
+
+ with reporter.step("Check put object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=OBJECT_ACCESS_DENIED):
+ put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check put object from container on the second node, expected allow"):
+ with expect_not_raises():
+ put_object(
+ default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint(), copies_number=3
+ )
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyPutObject",
+ )
+
+ with reporter.step("Check get object in container on the first node, expected allow"):
+ with expect_not_raises():
+ put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("LocalOverride: Deny to HeadObject in root tenant")
+ def test_local_override_deny_to_head_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 2",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyHeadObject",
+ rule=f"deny Object.Head /{cid}/*",
+ )
+
+ with reporter.step("Put object in container on the first node"):
+ oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check head object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HEAD_OBJECT)):
+ head_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check head object from container on the second node, expected allow"):
+ with expect_not_raises():
+ head_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyHeadObject",
+ )
+
+ with reporter.step("Check head object in container on the first node, expected allow"):
+ with expect_not_raises():
+ head_object(default_user.wallet, cid, oid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("LocalOverride: Deny to SearchObject in root tenant")
+ def test_local_override_deny_to_search_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 2",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denySearchObject",
+ rule=f"deny Object.Search /{cid}/*",
+ )
+
+ with reporter.step("Check search object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.SEARCH_OBJECT)):
+ search_object(default_user.wallet, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check search object from container on the second node, expected allow"):
+ with expect_not_raises():
+ search_object(default_user.wallet, cid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denySearchObject",
+ )
+
+ with reporter.step("Check search object in container on the first node, expected allow"):
+ with expect_not_raises():
+ search_object(default_user.wallet, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("LocalOverride: Deny to RangeObject in root tenant")
+ def test_local_override_deny_to_range_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 2",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyRangeObject",
+ rule=f"deny Object.Range /{cid}/*",
+ )
+
+ with reporter.step("Put object in container on the first node"):
+ oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check range object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.RANGE_OBJECT)):
+ get_range(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check get range object from container on the second node, expected allow"):
+ with expect_not_raises():
+ get_range(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyRangeObject",
+ )
+
+ with reporter.step("Check get range object in container on the first node, expected allow"):
+ with expect_not_raises():
+ get_range(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("LocalOverride: Deny to HashObject in root tenant")
+ def test_local_override_deny_to_hash_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 2",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyHashObject",
+ rule=f"deny Object.Hash /{cid}/*",
+ )
+
+ with reporter.step("Put object in container on the first node"):
+ oid = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check get range hash object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.HASH_OBJECT)):
+ get_range_hash(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check get range hash object from container on the second node, expected allow"):
+ with expect_not_raises():
+ get_range_hash(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyHashObject",
+ )
+
+ with reporter.step("Check get range hash object in container on the first node, expected allow"):
+ with expect_not_raises():
+ get_range_hash(default_user.wallet, cid, oid, "0:10", self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ @allure.title("LocalOverride: Deny to DeleteObject in root tenant")
+ def test_local_override_deny_to_delete_object_root(
+ self,
+ default_user: User,
+ remote_frostfs_cli_first_node: FrostfsCli,
+ frostfs_cli: FrostfsCli,
+ simple_object_size: ObjectSize,
+ ):
+
+ test_file = generate_file(simple_object_size.value)
+
+ with reporter.step("Create a container on the first node"):
+ cid = (
+ frostfs_cli.container.create(
+ rpc_endpoint=self.cluster.storage_nodes[0].get_rpc_endpoint(),
+ policy="REP 2",
+ name="dcl1",
+ await_mode=True,
+ basic_acl="public-read-write",
+ )
+ .stdout.split(" ")[1]
+ .strip()
+ .split("\n")[0]
+ )
+
+ with reporter.step("Create a container rule for the first node"):
+ remote_frostfs_cli_first_node.control.add_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyDeleteObject",
+ rule=f"deny Object.Delete /{cid}/*",
+ )
+
+ with reporter.step("Put objects in container on the first node"):
+ oid_1 = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ oid_2 = put_object(default_user.wallet, test_file, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Search object in container on the first node"):
+ search_object_in_container_1 = search_object(
+ default_user.wallet, cid, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint()
+ )
+ assert oid_1 in search_object_in_container_1, f"Object {oid_1} was not found"
+ assert oid_2 in search_object_in_container_1, f"Object {oid_2} was not found"
+
+ with reporter.step("Search object from container on the second node"):
+ search_object_in_container_2 = search_object(
+ default_user.wallet, cid, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint()
+ )
+ assert oid_1 in search_object_in_container_2, f"Object {oid_1} was not found"
+ assert oid_2 in search_object_in_container_2, f"Object {oid_2} was not found"
+
+ with reporter.step("Check delete object from container on the first node, expected access denied error"):
+ with pytest.raises(RuntimeError, match=RULE_ACCESS_DENIED_OBJECT.format(operation=Operations.DELETE_OBJECT)):
+ delete_object(default_user.wallet, cid, oid_1, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())
+
+ with reporter.step("Check delete object from container on the second node, expected allow"):
+ with expect_not_raises():
+ delete_object(default_user.wallet, cid, oid_2, self.shell, self.cluster.storage_nodes[1].get_rpc_endpoint())
+
+ with reporter.step("Delete a rule"):
+ remote_frostfs_cli_first_node.control.remove_rule(
+ endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
+ target_type="container",
+ target_name=f"{cid}",
+ chain_id="denyDeleteObject",
+ )
+
+ with reporter.step("Check delete object in container on the first node, expected allow"):
+ with expect_not_raises():
+ delete_object(default_user.wallet, cid, oid_1, self.shell, self.cluster.storage_nodes[0].get_rpc_endpoint())