src/frostfs_testlib/cli/frostfs_cli/ape_manager.py

@@ -0,0 +1,70 @@
+from typing import Optional
+
+from frostfs_testlib.cli.cli_command import CliCommand
+from frostfs_testlib.shell import CommandResult
+
+
+class FrostfsCliApeManager(CliCommand):
+    """Operations with APE manager."""
+
+    def add(
+        self,
+        rpc_endpoint: str,
+        chain_id: Optional[str] = None,
+        chain_id_hex: Optional[str] = None,
+        path: Optional[str] = None,
+        rule: Optional[str] | Optional[list[str]] = None,
+        target_name: Optional[str] = None,
+        target_type: Optional[str] = None,
+        wallet: Optional[str] = None,
+        address: Optional[str] = None,
+        timeout: Optional[str] = None,
+    ) -> CommandResult:
+        """Add rule chain for a target."""
+
+        return self._execute(
+            "ape-manager add",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )
+
+    def list(
+        self,
+        rpc_endpoint: str,
+        target_name: Optional[str] = None,
+        target_type: Optional[str] = None,
+        wallet: Optional[str] = None,
+        address: Optional[str] = None,
+        timeout: Optional[str] = None,
+    ) -> CommandResult:
+        """Generate APE override by target and APE chains. Util command.
+
+        Generated APE override can be dumped to a file in JSON format that is passed to
+        "create" command.
+        """
+
+        return self._execute(
+            "ape-manager list",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )
+
+    def remove(
+        self,
+        rpc_endpoint: str,
+        chain_id: Optional[str] = None,
+        chain_id_hex: Optional[str] = None,
+        target_name: Optional[str] = None,
+        target_type: Optional[str] = None,
+        wallet: Optional[str] = None,
+        address: Optional[str] = None,
+        timeout: Optional[str] = None,
+    ) -> CommandResult:
+        """Generate APE override by target and APE chains. Util command.
+
+        Generated APE override can be dumped to a file in JSON format that is passed to
+        "create" command.
+        """
+
+        return self._execute(
+            "ape-manager remove",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )

src/frostfs_testlib/cli/frostfs_cli/bearer.py

@@ -0,0 +1,54 @@
+from typing import Optional
+
+from frostfs_testlib.cli.cli_command import CliCommand
+from frostfs_testlib.shell import CommandResult
+
+
+class FrostfsCliBearer(CliCommand):
+    def create(
+        self,
+        rpc_endpoint: str,
+        out: str,
+        issued_at: Optional[str] = None,
+        expire_at: Optional[str] = None,
+        not_valid_before: Optional[str] = None,
+        ape: Optional[str] = None,
+        eacl: Optional[str] = None,
+        owner: Optional[str] = None,
+        json: Optional[bool] = False,
+        impersonate: Optional[bool] = False,
+        wallet: Optional[str] = None,
+        address: Optional[str] = None,
+    ) -> CommandResult:
+        """Create bearer token.
+
+        All epoch flags can be specified relative to the current epoch with the +n syntax.
+        In this case --rpc-endpoint flag should be specified and the epoch in bearer token
+        is set to current epoch + n.
+        """
+        return self._execute(
+            "bearer create",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )
+
+    def generate_ape_override(
+        self,
+        chain_id: Optional[str] = None,
+        chain_id_hex: Optional[str] = None,
+        cid: Optional[str] = None,
+        output: Optional[str] = None,
+        path: Optional[str] = None,
+        rule: Optional[str] = None,
+        wallet: Optional[str] = None,
+        address: Optional[str] = None,
+    ) -> CommandResult:
+        """Generate APE override by target and APE chains. Util command.
+
+        Generated APE override can be dumped to a file in JSON format that is passed to
+        "create" command.
+        """
+
+        return self._execute(
+            "bearer generate-ape-override",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )

src/frostfs_testlib/cli/frostfs_cli/cli.py

@@ -2,6 +2,8 @@ from typing import Optional
 
 from frostfs_testlib.cli.frostfs_cli.accounting import FrostfsCliAccounting
 from frostfs_testlib.cli.frostfs_cli.acl import FrostfsCliACL
+from frostfs_testlib.cli.frostfs_cli.ape_manager import FrostfsCliApeManager
+from frostfs_testlib.cli.frostfs_cli.bearer import FrostfsCliBearer
 from frostfs_testlib.cli.frostfs_cli.container import FrostfsCliContainer
 from frostfs_testlib.cli.frostfs_cli.control import FrostfsCliControl
 from frostfs_testlib.cli.frostfs_cli.netmap import FrostfsCliNetmap
@@ -41,3 +43,5 @@ class FrostfsCli:
         self.version = FrostfsCliVersion(shell, frostfs_cli_exec_path, config=config_file)
         self.tree = FrostfsCliTree(shell, frostfs_cli_exec_path, config=config_file)
         self.control = FrostfsCliControl(shell, frostfs_cli_exec_path, config=config_file)
+        self.bearer = FrostfsCliBearer(shell, frostfs_cli_exec_path, config=config_file)
+        self.ape_manager = FrostfsCliApeManager(shell, frostfs_cli_exec_path, config=config_file)

src/frostfs_testlib/cli/frostfs_cli/util.py

@@ -54,3 +54,11 @@ class FrostfsCliUtil(CliCommand):
             "util sign session-token",
             **{param: value for param, value in locals().items() if param not in ["self"]},
         )
+
+    def convert_eacl(self, from_file: str, to_file: str, json: Optional[bool] = False, ape: Optional[bool] = False):
+        """Convert representation of extended ACL table."""
+
+        return self._execute(
+            "util convert eacl",
+            **{param: value for param, value in locals().items() if param not in ["self"]},
+        )

src/frostfs_testlib/storage/dataclasses/ape.py

@@ -0,0 +1,115 @@
+import logging
+from dataclasses import dataclass
+from enum import Enum
+from typing import Optional
+
+from frostfs_testlib.testing.readable import HumanReadableEnum
+from frostfs_testlib.utils import string_utils
+
+logger = logging.getLogger("NeoLogger")
+EACL_LIFETIME = 100500
+FROSTFS_CONTRACT_CACHE_TIMEOUT = 30
+
+
+class ObjectOperations(HumanReadableEnum):
+    PUT = "object.put"
+    GET = "object.get"
+    HEAD = "object.head"
+    GET_RANGE = "object.range"
+    GET_RANGE_HASH = "object.hash"
+    SEARCH = "object.search"
+    DELETE = "object.delete"
+    WILDCARD_ALL = "object.*"
+
+    @staticmethod
+    def get_all():
+        return [op for op in ObjectOperations if op != ObjectOperations.WILDCARD_ALL]
+
+
+class Verb(HumanReadableEnum):
+    ALLOW = "allow"
+    DENY = "deny"
+
+
+class Role(HumanReadableEnum):
+    OWNER = "owner"
+    IR = "ir"
+    CONTAINER = "container"
+    OTHERS = "others"
+
+
+class ConditionType(HumanReadableEnum):
+    RESOURCE = "ResourceCondition"
+    REQUEST = "RequestCondition"
+
+
+# See https://git.frostfs.info/TrueCloudLab/policy-engine/src/branch/master/schema/native/consts.go#L40-L53
+class ConditionKey(HumanReadableEnum):
+    ROLE = '"\\$Actor:role"'
+    PUBLIC_KEY = '"\\$Actor:publicKey"'
+
+
+class MatchType(HumanReadableEnum):
+    EQUAL = "="
+    NOT_EQUAL = "!="
+
+
+@dataclass
+class Condition:
+    condition_key: ConditionKey | str
+    condition_value: str
+    condition_type: ConditionType = ConditionType.REQUEST
+    match_type: MatchType = MatchType.EQUAL
+
+    def as_string(self):
+        key = self.condition_key.value if isinstance(self.condition_key, ConditionKey) else self.condition_key
+        value = self.condition_value.value if isinstance(self.condition_value, Enum) else self.condition_value
+
+        return f"{self.condition_type.value}:{key}{self.match_type.value}{value}"
+
+    @staticmethod
+    def by_role(*args, **kwargs) -> "Condition":
+        return Condition(ConditionKey.ROLE, *args, **kwargs)
+
+    @staticmethod
+    def by_key(*args, **kwargs) -> "Condition":
+        return Condition(ConditionKey.PUBLIC_KEY, *args, **kwargs)
+
+
+class Rule:
+    def __init__(
+        self,
+        access: Verb,
+        operations: list[ObjectOperations] | ObjectOperations,
+        conditions: list[Condition] | Condition = None,
+        chain_id: Optional[str] = None,
+    ) -> None:
+        self.access = access
+        self.operations = operations
+
+        if not conditions:
+            self.conditions = []
+        elif isinstance(conditions, Condition):
+            self.conditions = [conditions]
+        else:
+            self.conditions = conditions
+
+        if not isinstance(self.conditions, list):
+            raise RuntimeError("Conditions must be a list")
+
+        if not operations:
+            self.operations = []
+        elif isinstance(operations, ObjectOperations):
+            self.operations = [operations]
+        else:
+            self.operations = operations
+
+        if not isinstance(self.operations, list):
+            raise RuntimeError("Operations must be a list")
+
+        self.chain_id = chain_id if chain_id else string_utils.unique_name("chain-id-")
+
+    def as_string(self):
+        conditions = " ".join([cond.as_string() for cond in self.conditions])
+        operations = " ".join([op.value for op in self.operations])
+        return f"{self.access.value} {operations} {conditions} *"

src/frostfs_testlib/testing/cluster_test_base.py

@@ -32,7 +32,7 @@ class ClusterTestBase:
     ):
         epoch.tick_epoch(self.shell, self.cluster, alive_node=alive_node)
         if wait_block:
-            time.sleep(datetime_utils.parse_time(MORPH_BLOCK_TIME) * wait_block)
+            self.wait_for_blocks(wait_block)
 
     def wait_for_epochs_align(self):
         epoch.wait_for_epochs_align(self.shell, self.cluster)
@@ -42,3 +42,6 @@ class ClusterTestBase:
 
     def ensure_fresh_epoch(self):
         return epoch.ensure_fresh_epoch(self.shell, self.cluster)
+
+    def wait_for_blocks(self, blocks_count: int = 1):
+        time.sleep(datetime_utils.parse_time(MORPH_BLOCK_TIME) * blocks_count)

src/frostfs_testlib/utils/string_utils.py

@@ -8,7 +8,7 @@ DIGITS_AND_ASCII_LETTERS = string.ascii_letters + string.digits
 NON_DIGITS_AND_LETTERS = string.punctuation
 
 
-def unique_name(prefix: str = ""):
+def unique_name(prefix: str = "", postfix: str = ""):
     """
     Generate unique short name of anything with prefix.
     This should be unique in scope of multiple runs
@@ -18,7 +18,7 @@ def unique_name(prefix: str = ""):
     Returns:
         unique name string
     """
-    return f"{prefix}{hex(int(datetime.now().timestamp() * 1000000))}"
+    return f"{prefix}{hex(int(datetime.now().timestamp() * 1000000))}{postfix}"
 
 
 def random_string(length: int = 5, source: str = ONLY_ASCII_LETTERS):