frostfs-testcases/pytest_tests/testsuites/access/ape/test_ape.py

import allure
import pytest
from frostfs_testlib import reporter
from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
from frostfs_testlib.resources.wellknown_acl import PUBLIC_ACL
from frostfs_testlib.steps.cli.object import put_object_to_random_node
from frostfs_testlib.steps.node_management import drop_object
from frostfs_testlib.storage.dataclasses import ape
from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
from frostfs_testlib.utils import wallet_utils
from frostfs_testlib.utils.failover_utils import wait_object_replication
from frostfs_testlib.utils.file_utils import TestFile

from ....helpers.container_access import (
    ALL_OBJECT_OPERATIONS,
    assert_access_to_container,
    assert_full_access_to_container,
    assert_no_access_to_container,
)
from ....helpers.container_spec import ContainerSpec


@pytest.fixture
def denied_wallet(default_wallet: WalletInfo, other_wallet: WalletInfo, role: ape.Role) -> WalletInfo:
    return other_wallet if role == ape.Role.OTHERS else default_wallet


@pytest.fixture
def allowed_wallet(default_wallet: WalletInfo, other_wallet: WalletInfo, role: ape.Role) -> WalletInfo:
    return default_wallet if role == ape.Role.OTHERS else other_wallet


@pytest.mark.nightly
@pytest.mark.ape
class TestApeContainer(ClusterTestBase):
    @pytest.mark.sanity
    @allure.title("Deny operations via APE by role (role={role}, obj_size={object_size})")
    @pytest.mark.parametrize("role", [ape.Role.OWNER, ape.Role.OTHERS], indirect=True)
    def test_deny_operations_via_ape_by_role(
        self,
        denied_wallet: WalletInfo,
        allowed_wallet: WalletInfo,
        frostfs_cli: FrostfsCli,
        container: str,
        objects: list[str],
        role: ape.Role,
        file_path: TestFile,
        rpc_endpoint: str,
    ):
        with reporter.step(f"Deny all operations for {role} via APE"):
            deny_rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, ape.Condition.by_role(role.value))
            frostfs_cli.ape_manager.add(
                rpc_endpoint, deny_rule.chain_id, target_name=container, target_type="container", rule=deny_rule.as_string()
            )

        with reporter.step("Wait for one block"):
            self.wait_for_blocks()

        with reporter.step(f"Assert denied role have no access to public container"):
            # access checks will try to remove object, so we use .pop() to ensure we have object before deletion
            assert_no_access_to_container(denied_wallet, container, objects.pop(), file_path, self.shell, self.cluster)

        with reporter.step(f"Assert allowed role have full access to public container"):
            assert_full_access_to_container(allowed_wallet, container, objects.pop(), file_path, self.shell, self.cluster)

        with reporter.step(f"Remove deny rule from APE"):
            frostfs_cli.ape_manager.remove(rpc_endpoint, deny_rule.chain_id, target_name=container, target_type="container")

        with reporter.step("Wait for one block"):
            self.wait_for_blocks()

        with reporter.step("Assert allowed role have full access to public container"):
            assert_full_access_to_container(allowed_wallet, container, objects.pop(), file_path, self.shell, self.cluster)

        with reporter.step("Assert denied role have full access to public container"):
            assert_full_access_to_container(denied_wallet, container, objects.pop(), file_path, self.shell, self.cluster)

    @allure.title("Deny operations for others via APE excluding single pubkey (obj_size={object_size})")
    def test_deny_opeartions_excluding_pubkey(
        self,
        frostfs_cli: FrostfsCli,
        default_wallet: WalletInfo,
        other_wallet: WalletInfo,
        other_wallet_2: WalletInfo,
        container: str,
        objects: list[str],
        rpc_endpoint: str,
        file_path: TestFile,
    ):
        with reporter.step("Add deny APE rules for others except single wallet"):
            rule_conditions = [
                ape.Condition.by_role(ape.Role.OTHERS),
                ape.Condition.by_key(
                    wallet_utils.get_wallet_public_key(other_wallet_2.path, other_wallet_2.password),
                    match_type=ape.MatchType.NOT_EQUAL,
                ),
            ]
            rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, rule_conditions)
            frostfs_cli.ape_manager.add(rpc_endpoint, rule.chain_id, target_name=container, target_type="container", rule=rule.as_string())

        with reporter.step("Wait for one block"):
            self.wait_for_blocks()

        with reporter.step("Assert others have no access to public container"):
            # access checks will try to remove object, so we use .pop() to ensure we have object before deletion
            assert_no_access_to_container(other_wallet, container, objects[0], file_path, self.shell, self.cluster)

        with reporter.step("Assert owner have full access to public container"):
            assert_full_access_to_container(default_wallet, container, objects.pop(), file_path, self.shell, self.cluster)

        with reporter.step("Assert allowed wallet have full access to public container"):
            assert_full_access_to_container(other_wallet_2, container, objects.pop(), file_path, self.shell, self.cluster)

    @allure.title("Replication works with APE deny rules on OWNER and OTHERS (obj_size={object_size})")
    @pytest.mark.container(ContainerSpec(f"REP %NODE_COUNT% IN X CBF 1 SELECT %NODE_COUNT% FROM * AS X", PUBLIC_ACL))
    def test_replication_works_with_deny_rules(
        self,
        default_wallet: WalletInfo,
        frostfs_cli: FrostfsCli,
        container: str,
        rpc_endpoint: str,
        file_path: TestFile,
    ):
        with reporter.step("Put object to container"):
            oid = put_object_to_random_node(default_wallet, file_path, container, self.shell, self.cluster)

        with reporter.step("Wait for object replication after upload"):
            wait_object_replication(container, oid, len(self.cluster.cluster_nodes), self.shell, self.cluster.storage_nodes)

        with reporter.step("Add deny APE rules for owner and others"):
            rule_conditions = [
                ape.Condition.by_role(ape.Role.OWNER),
                ape.Condition.by_role(ape.Role.OTHERS),
            ]
            for rule_condition in rule_conditions:
                rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, rule_condition)
                frostfs_cli.ape_manager.add(
                    rpc_endpoint, rule.chain_id, target_name=container, target_type="container", rule=rule.as_string()
                )

        with reporter.step("Wait for one block"):
            self.wait_for_blocks()

        with reporter.step("Drop object"):
            drop_object(self.cluster.storage_nodes[0], container, oid)

        with reporter.step("Wait for dropped object to be replicated"):
            wait_object_replication(container, oid, len(self.cluster.storage_nodes), self.shell, self.cluster.storage_nodes)

    @allure.title("Deny operations via APE by role (role=ir, obj_size={object_size})")
    def test_deny_operations_via_ape_by_role_ir(
        self, frostfs_cli: FrostfsCli, ir_wallet: WalletInfo, container: str, objects: list[str], rpc_endpoint: str, file_path: TestFile
    ):
        default_ir_access = {
            ape.ObjectOperations.PUT: False,
            ape.ObjectOperations.GET: True,
            ape.ObjectOperations.HEAD: True,
            ape.ObjectOperations.GET_RANGE: False,
            ape.ObjectOperations.GET_RANGE_HASH: True,
            ape.ObjectOperations.SEARCH: True,
            ape.ObjectOperations.DELETE: False,
        }

        with reporter.step("Assert IR wallet access in default state"):
            assert_access_to_container(default_ir_access, ir_wallet, container, objects[0], file_path, self.shell, self.cluster)

        with reporter.step("Add deny APE rule with deny all operations for IR role"):
            rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, [ape.Condition.by_role(ape.Role.IR.value)])
            frostfs_cli.ape_manager.add(rpc_endpoint, rule.chain_id, target_name=container, target_type="container", rule=rule.as_string())

        with reporter.step("Wait for one block"):
            self.wait_for_blocks()

        with reporter.step("Assert IR wallet ignores APE rules"):
            assert_access_to_container(default_ir_access, ir_wallet, container, objects[0], file_path, self.shell, self.cluster)

        with reporter.step("Remove APE rule"):
            frostfs_cli.ape_manager.remove(rpc_endpoint, rule.chain_id, target_name=container, target_type="container")

        with reporter.step("Wait for one block"):
            self.wait_for_blocks()

        with reporter.step("Assert IR wallet access is restored"):
            assert_access_to_container(default_ir_access, ir_wallet, container, objects[0], file_path, self.shell, self.cluster)

    @allure.title("Deny operations via APE by role (role=container, obj_size={object_size})")
    def test_deny_operations_via_ape_by_role_container(
        self,
        frostfs_cli: FrostfsCli,
        container_node_wallet: WalletInfo,
        container: str,
        objects: list[str],
        rpc_endpoint: str,
        file_path: TestFile,
    ):
        access_matrix = {
            ape.ObjectOperations.PUT: True,
            ape.ObjectOperations.GET: True,
            ape.ObjectOperations.HEAD: True,
            ape.ObjectOperations.GET_RANGE: False,
            ape.ObjectOperations.GET_RANGE_HASH: True,
            ape.ObjectOperations.SEARCH: True,
            ape.ObjectOperations.DELETE: False,
        }

        with reporter.step("Assert CONTAINER wallet access in default state"):
            assert_access_to_container(access_matrix, container_node_wallet, container, objects[0], file_path, self.shell, self.cluster)

        rule = ape.Rule(ape.Verb.DENY, ALL_OBJECT_OPERATIONS, ape.Condition.by_role(ape.Role.CONTAINER.value))

        with reporter.step(f"Add APE rule with deny all operations for CONTAINER and IR roles"):
            frostfs_cli.ape_manager.add(rpc_endpoint, rule.chain_id, target_name=container, target_type="container", rule=rule.as_string())

        with reporter.step("Wait for one block"):
            self.wait_for_blocks()

        with reporter.step("Assert CONTAINER wallet ignores APE rule"):
            assert_access_to_container(access_matrix, container_node_wallet, container, objects[0], file_path, self.shell, self.cluster)

        with reporter.step("Remove APE rule"):
            frostfs_cli.ape_manager.remove(rpc_endpoint, rule.chain_id, target_name=container, target_type="container")

        with reporter.step("Wait for one block"):
            self.wait_for_blocks()

        with reporter.step("Assert CONTAINER wallet access after rule was removed"):
            assert_access_to_container(access_matrix, container_node_wallet, container, objects[0], file_path, self.shell, self.cluster)