[#188] Add CredentialsProvider

Signed-off-by: Andrey Berezin <a.berezin@yadro.com>

src/frostfs_testlib/credentials/authmate_s3.py

@@ -0,0 +1,49 @@
+import re
+from datetime import datetime
+
+from frostfs_testlib import reporter
+from frostfs_testlib.cli import FrostfsAuthmate
+from frostfs_testlib.credentials.interfaces import S3CredentialsProvider
+from frostfs_testlib.resources.cli import FROSTFS_AUTHMATE_EXEC
+from frostfs_testlib.shell import Shell
+from frostfs_testlib.steps.cli.container import list_containers
+from frostfs_testlib.storage.cluster import Cluster, ClusterNode
+from frostfs_testlib.storage.dataclasses.wallet import WalletInfo
+
+
+class AuthmateS3CredentialsProvider(S3CredentialsProvider):
+    @reporter.step("Init S3 Credentials using Authmate CLI")
+    def provide(self, cluster_node: ClusterNode) -> tuple[str, str]:
+        cluster: Cluster = self.stash["cluster"]
+        shell: Shell = self.stash["shell"]
+        wallet: WalletInfo = self.stash["wallet"]
+        endpoint = cluster_node.storage_node.get_rpc_endpoint()
+
+        gate_public_keys = [s3gate.get_wallet_public_key() for s3gate in cluster.s3_gates]
+        # 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=self.stash.get("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.path, shell, endpoint)
+        assert cid in containers_list, f"Expected cid {cid} in {containers_list}"
+
+        return aws_access_key_id, aws_secret_access_key

src/frostfs_testlib/credentials/interfaces.py

@@ -0,0 +1,25 @@
+from abc import abstractmethod
+
+from frostfs_testlib.plugins import load_plugin
+from frostfs_testlib.storage.cluster import ClusterNode
+
+
+class S3CredentialsProvider(object):
+    stash: dict
+
+    def __init__(self, stash: dict) -> None:
+        self.stash = stash
+
+    @abstractmethod
+    def provide(self, cluster_node: ClusterNode) -> tuple[str, str]:
+        raise NotImplementedError("Directly called abstract class?")
+
+
+class CredentialsProvider(object):
+    stash: dict
+    S3: S3CredentialsProvider
+
+    def __init__(self, s3_plugin_name: str) -> None:
+        self.stash = {}
+        s3cls = load_plugin("frostfs.testlib.credentials_providers", s3_plugin_name)
+        self.S3 = s3cls(self.stash)