forked from TrueCloudLab/frostfs-testlib
47 lines
2.3 KiB
Python
47 lines
2.3 KiB
Python
import re
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from frostfs_testlib import reporter
|
|
from frostfs_testlib.cli import FrostfsAuthmate
|
|
from frostfs_testlib.credentials.interfaces import S3Credentials, S3CredentialsProvider, User
|
|
from frostfs_testlib.resources.cli import FROSTFS_AUTHMATE_EXEC
|
|
from frostfs_testlib.shell import LocalShell
|
|
from frostfs_testlib.steps.cli.container import list_containers
|
|
from frostfs_testlib.storage.cluster import ClusterNode
|
|
from frostfs_testlib.storage.dataclasses.frostfs_services import S3Gate
|
|
|
|
|
|
class AuthmateS3CredentialsProvider(S3CredentialsProvider):
|
|
@reporter.step("Init S3 Credentials using Authmate CLI")
|
|
def provide(self, user: User, cluster_node: ClusterNode, location_constraints: Optional[str] = None) -> S3Credentials:
|
|
cluster_nodes: list[ClusterNode] = self.cluster.cluster_nodes
|
|
shell = LocalShell()
|
|
wallet = user.wallet
|
|
endpoint = cluster_node.storage_node.get_rpc_endpoint()
|
|
|
|
gate_public_keys = [node.service(S3Gate).get_wallet_public_key() for node in cluster_nodes]
|
|
# unique short bucket name
|
|
bucket = f"bucket-{hex(int(datetime.now().timestamp()*1000000))}"
|
|
|
|
frostfs_authmate: FrostfsAuthmate = FrostfsAuthmate(shell, FROSTFS_AUTHMATE_EXEC)
|
|
issue_secret_output = frostfs_authmate.secret.issue(
|
|
wallet=wallet.path,
|
|
peer=endpoint,
|
|
gate_public_key=gate_public_keys,
|
|
wallet_password=wallet.password,
|
|
container_policy=location_constraints,
|
|
container_friendly_name=bucket,
|
|
).stdout
|
|
|
|
aws_access_key_id = str(re.search(r"access_key_id.*:\s.(?P<aws_access_key_id>\w*)", issue_secret_output).group("aws_access_key_id"))
|
|
aws_secret_access_key = str(
|
|
re.search(r"secret_access_key.*:\s.(?P<aws_secret_access_key>\w*)", issue_secret_output).group("aws_secret_access_key")
|
|
)
|
|
cid = str(re.search(r"container_id.*:\s.(?P<container_id>\w*)", issue_secret_output).group("container_id"))
|
|
|
|
containers_list = list_containers(wallet, shell, endpoint)
|
|
assert cid in containers_list, f"Expected cid {cid} in {containers_list}"
|
|
|
|
user.s3_credentials = S3Credentials(aws_access_key_id, aws_secret_access_key)
|
|
return user.s3_credentials
|