frostfs-testcases/pytest_tests/testsuites/services/http_gate/test_http_bearer.py

import logging

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.acl import bearer_token_base64_from_file
from frostfs_testlib.steps.cli.container import create_container
from frostfs_testlib.steps.http.http_gate import upload_via_http_gate_curl, verify_object_hash
from frostfs_testlib.storage.cluster import Cluster
from frostfs_testlib.storage.dataclasses import ape
from frostfs_testlib.storage.dataclasses.object_size import ObjectSize
from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
from frostfs_testlib.testing.cluster_test_base import ClusterTestBase
from frostfs_testlib.utils.file_utils import generate_file

from ....helpers.bearer_token import create_bearer_token

logger = logging.getLogger("NeoLogger")


@pytest.mark.http_gate
@pytest.mark.http_put
class Test_http_bearer(ClusterTestBase):
    PLACEMENT_RULE = "REP 2 IN X CBF 1 SELECT 2 FROM * AS X"

    @pytest.fixture(scope="class")
    def user_container(self, frostfs_cli: FrostfsCli, default_wallet: WalletInfo, cluster: Cluster) -> str:
        with reporter.step("Create container"):
            cid = create_container(default_wallet, self.shell, self.cluster.default_rpc_endpoint, self.PLACEMENT_RULE, PUBLIC_ACL)

        with reporter.step("Deny PUT via APE rule to container"):
            role_condition = ape.Condition.by_role(ape.Role.OWNER)
            rule = ape.Rule(ape.Verb.DENY, ape.ObjectOperations.PUT, role_condition)
            frostfs_cli.ape_manager.add(
                cluster.default_rpc_endpoint, rule.chain_id, target_name=cid, target_type="container", rule=rule.as_string()
            )

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

        return cid

    @pytest.fixture(scope="class")
    def bearer_token(self, frostfs_cli: FrostfsCli, user_container: str, temp_directory: str, cluster: Cluster) -> str:
        with reporter.step(f"Create bearer token for {ape.Role.OTHERS} with all operations allowed"):
            role_condition = ape.Condition.by_role(ape.Role.OTHERS)
            rule = ape.Rule(ape.Verb.ALLOW, ape.ObjectOperations.WILDCARD_ALL, role_condition)
            bearer = create_bearer_token(frostfs_cli, temp_directory, user_container, rule, cluster.default_rpc_endpoint)

            return bearer_token_base64_from_file(bearer)

    @allure.title(f"[NEGATIVE] Put object without bearer token for {ape.Role.OTHERS}")
    def test_unable_put_without_bearer_token(self, simple_object_size: ObjectSize, user_container: str):
        upload_via_http_gate_curl(
            cid=user_container,
            filepath=generate_file(simple_object_size.value),
            endpoint=self.cluster.default_http_gate_endpoint,
            error_pattern="access to object operation denied",
        )

    @allure.title("Put object via HTTP using bearer token (object_size={object_size})")
    def test_put_with_bearer_when_eacl_restrict(
        self,
        object_size: ObjectSize,
        default_wallet: WalletInfo,
        user_container: str,
        bearer_token: str,
    ):
        file_path = generate_file(object_size.value)
        with reporter.step(f"Put object with bearer token for {ape.Role.OTHERS}, then get and verify hashes"):
            headers = [f" -H 'Authorization: Bearer {bearer_token}'"]
            oid = upload_via_http_gate_curl(
                cid=user_container,
                filepath=file_path,
                endpoint=self.cluster.default_http_gate_endpoint,
                headers=headers,
            )
            verify_object_hash(
                oid=oid,
                file_name=file_path,
                wallet=default_wallet,
                cid=user_container,
                shell=self.shell,
                nodes=self.cluster.storage_nodes,
                request_node=self.cluster.cluster_nodes[0],
            )