a.berezin
b999d7cf9b
Some checks failed
DCO check / DCO (pull_request) Has been cancelled
Signed-off-by: a.berezin <a.berezin@yadro.com>
75 lines
3.3 KiB
Python
75 lines
3.3 KiB
Python
import logging
|
|
|
|
import allure
|
|
import pytest
|
|
from frostfs_testlib import reporter
|
|
from frostfs_testlib.cli.frostfs_cli.cli import FrostfsCli
|
|
from frostfs_testlib.steps.acl import bearer_token_base64_from_file
|
|
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
|
|
from ....helpers.container_request import APE_EVERYONE_ALLOW_ALL, ContainerRequest, requires_container
|
|
|
|
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"
|
|
OWNER_ROLE = ape.Condition.by_role(ape.Role.OWNER)
|
|
CUSTOM_APE_RULE = ape.Rule(ape.Verb.DENY, ape.ObjectOperations.PUT, OWNER_ROLE)
|
|
|
|
@pytest.fixture()
|
|
def bearer_token(self, frostfs_cli: FrostfsCli, 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, 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, container: str):
|
|
upload_via_http_gate_curl(
|
|
cid=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})")
|
|
@requires_container(
|
|
ContainerRequest(PLACEMENT_RULE, [APE_EVERYONE_ALLOW_ALL, CUSTOM_APE_RULE], short_name="custom with denied owner put")
|
|
)
|
|
def test_put_with_bearer_when_ape_restrict(
|
|
self,
|
|
object_size: ObjectSize,
|
|
default_wallet: WalletInfo,
|
|
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=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=container,
|
|
shell=self.shell,
|
|
nodes=self.cluster.storage_nodes,
|
|
request_node=self.cluster.cluster_nodes[0],
|
|
)
|