frostfs-testcases/pytest_tests/testsuites/ape/test_ape_local_object_allow.py

import allure
import pytest
from frostfs_testlib import reporter
from frostfs_testlib.cli import FrostfsCli
from frostfs_testlib.resources.error_patterns import NO_RULE_FOUND_OBJECT
from frostfs_testlib.storage.grpc_operations.interfaces import GrpcClientWrapper
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
from frostfs_testlib.testing.test_control import expect_not_raises
from frostfs_testlib.utils.file_utils import TestFile

from ...helpers.container_request import ContainerRequest

REP1_MSK = ContainerRequest("REP 1 IN MOW CBF 1 SELECT 1 FROM MSK AS MOW FILTER SubDivCode EQ MOW AS MSK", short_name="REP1_MSK_no_ape")


@pytest.mark.ape
@pytest.mark.ape_local
@pytest.mark.ape_object
@pytest.mark.ape_allow
@pytest.mark.parametrize("container_request", [REP1_MSK], indirect=True)
@pytest.mark.parametrize("user_tag", ["ApeLocalOverrideAllow"], indirect=True)  # provide dedicated user with no APE side-policies
class TestApeLocalOverrideAllow(ClusterTestBase):
    @allure.title("LocalOverride: Allow to GetObject in root tenant")
    def test_local_override_allow_to_get_object_root(
        self,
        frostfs_cli_on_first_node: FrostfsCli,
        grpc_client: GrpcClientWrapper,
        container: str,
        object_id: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowGetObject",
                rule=f"allow Object.Get *",
            )

        with reporter.step("Check get object in container on the first node, expected allow"):
            with expect_not_raises():
                grpc_client.object.get(container, object_id, self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Check get object in container on the second node, epxected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                grpc_client.object.get(container, object_id, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowGetObject",
            )

    @allure.title("LocalOverride: Allow to PutObject in root tenant")
    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
    def test_local_override_allow_to_put_object_root(
        self,
        frostfs_cli_on_first_node: FrostfsCli,
        grpc_client: GrpcClientWrapper,
        container: str,
        test_file: TestFile,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowPutObject",
                rule=f"allow Object.Put *",
            )

        with reporter.step("Check put object in container on the first node, expected allow"):
            with expect_not_raises():
                grpc_client.object.put(test_file, container, self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Check put object in container on the second node, epxected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                grpc_client.object.put(test_file, container, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowPutObject",
            )

    @allure.title("LocalOverride: Allow to HeadObject in root tenant")
    def test_local_override_allow_to_head_object_root(
        self,
        frostfs_cli_on_first_node: FrostfsCli,
        grpc_client: GrpcClientWrapper,
        container: str,
        object_id: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowHeadObject",
                rule=f"allow Object.Head *",
            )

        with reporter.step("Check head object in container on the first node, expected allow"):
            with expect_not_raises():
                grpc_client.object.head(container, object_id, self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Check head object in container on the second node, expected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                grpc_client.object.head(container, object_id, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowHeadObject",
            )

    @allure.title("LocalOverride: Allow to SearchObject in root tenant")
    def test_local_override_allow_to_search_object_root(
        self,
        frostfs_cli_on_first_node: FrostfsCli,
        grpc_client: GrpcClientWrapper,
        container: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowSearchObject",
                rule=f"allow Object.Search *",
            )

        with reporter.step("Check search object in container on the first node, expected allow"):
            with expect_not_raises():
                grpc_client.object.search(container, 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=NO_RULE_FOUND_OBJECT):
                grpc_client.object.search(container, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowSearchObject",
            )

    @allure.title("LocalOverride: Allow to RangeObject in root tenant")
    def test_local_override_allow_to_range_object_root(
        self,
        frostfs_cli_on_first_node: FrostfsCli,
        grpc_client: GrpcClientWrapper,
        container: str,
        object_id: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowRangeObject",
                rule=f"allow Object.Range *",
            )

        with reporter.step("Check get range object in container on the first node, expected allow"):
            with expect_not_raises():
                grpc_client.object.range(container, object_id, "0:10", self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Check get range object in container on the second node, expected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                grpc_client.object.range(container, object_id, "0:10", self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowRangeObject",
            )

    @allure.title("LocalOverride: Allow to HashObject in root tenant")
    def test_local_override_allow_to_hash_object_root(
        self,
        frostfs_cli_on_first_node: FrostfsCli,
        grpc_client: GrpcClientWrapper,
        container: str,
        object_id: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowHashObject",
                rule=f"allow Object.Hash *",
            )

        with reporter.step("Check get range hash object in container on the first node, expected allow"):
            with expect_not_raises():
                grpc_client.object.hash(self.cluster.storage_nodes[0].get_rpc_endpoint(), container, object_id, range="0:10")

        with reporter.step("Check get range hash object in container on the second node, expected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                grpc_client.object.hash(self.cluster.storage_nodes[1].get_rpc_endpoint(), container, object_id, range="0:10")

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowHashObject",
            )

    @allure.title("LocalOverride: Allow to DeleteObject in root tenant")
    def test_local_override_allow_to_delete_object_root(
        self,
        frostfs_cli_on_first_node: FrostfsCli,
        grpc_client: GrpcClientWrapper,
        container: str,
        object_id: str,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowDeleteObject",
                rule=f"allow Object.Head Object.Delete *",
            )

        with reporter.step("Check delete object from container on the second node, expected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                grpc_client.object.delete(container, object_id, self.cluster.storage_nodes[1].get_rpc_endpoint())

        with reporter.step("Check delete object in container on the first node, expected allow"):
            with expect_not_raises():
                grpc_client.object.delete(container, object_id, self.cluster.storage_nodes[0].get_rpc_endpoint())

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowDeleteObject",
            )

    @allure.title("LocalOverride: Allow to PatchObject in root tenant")
    @pytest.mark.parametrize("object_size", ["simple"], indirect=True)
    def test_local_override_allow_to_patch_object_root(
        self,
        frostfs_cli_on_first_node: FrostfsCli,
        grpc_client: GrpcClientWrapper,
        container: str,
        object_id: str,
        test_file: TestFile,
    ):
        with reporter.step("Create local override on first node"):
            frostfs_cli_on_first_node.control.add_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowPatchObject",
                rule=f"allow Object.Patch *",
            )

        with reporter.step("Check patch object in container on the second node, epxected access denied error"):
            with pytest.raises(RuntimeError, match=NO_RULE_FOUND_OBJECT):
                grpc_client.object.patch(
                    container,
                    object_id,
                    self.cluster.storage_nodes[1].get_rpc_endpoint(),
                    ranges=["500:300"],
                    payloads=[test_file],
                    new_attrs="patched=false",
                    timeout="200s",
                )

        with reporter.step("Check patch object in container on the first node, expected allow"):
            with expect_not_raises():
                patched_oid = grpc_client.object.patch(
                    container,
                    object_id,
                    self.cluster.storage_nodes[0].get_rpc_endpoint(),
                    ranges=["100:200"],
                    payloads=[test_file],
                    new_attrs="patched=true",
                    timeout="200s",
                )
                assert patched_oid != object_id, "OID of patched object must be different from original one"

        with reporter.step("Delete a rule"):
            frostfs_cli_on_first_node.control.remove_rule(
                endpoint=self.cluster.storage_nodes[0].get_control_endpoint(),
                target_type="container",
                target_name=container,
                chain_id="allowPatchObject",
            )