Bitbucket server, WIP

pr_agent/secret_providers/__init__.py

@@ -0,0 +1,16 @@
+from pr_agent.config_loader import get_settings
+
+
+def get_secret_provider():
+    try:
+        provider_id = get_settings().config.secret_provider
+    except AttributeError as e:
+        raise ValueError("secret_provider is a required attribute in the configuration file") from e
+    try:
+        if provider_id == 'google_cloud_storage':
+            from pr_agent.secret_providers.google_cloud_storage_secret_provider import GoogleCloudStorageSecretProvider
+            return GoogleCloudStorageSecretProvider()
+        else:
+            raise ValueError(f"Unknown secret provider: {provider_id}")
+    except Exception as e:
+        raise ValueError(f"Failed to initialize secret provider {provider_id}") from e

pr_agent/secret_providers/google_cloud_storage_secret_provider.py

@@ -0,0 +1,35 @@
+import ujson
+
+from google.cloud import storage
+
+from pr_agent.config_loader import get_settings
+from pr_agent.git_providers.gitlab_provider import logger
+from pr_agent.secret_providers.secret_provider import SecretProvider
+
+
+class GoogleCloudStorageSecretProvider(SecretProvider):
+    def __init__(self):
+        try:
+            self.client = storage.Client.from_service_account_info(ujson.loads(get_settings().google_cloud_storage.
+                                                                               service_account))
+            self.bucket_name = get_settings().google_cloud_storage.bucket_name
+            self.bucket = self.client.bucket(self.bucket_name)
+        except Exception as e:
+            logger.error(f"Failed to initialize Google Cloud Storage Secret Provider: {e}")
+            raise e
+
+    def get_secret(self, secret_name: str) -> str:
+        try:
+            blob = self.bucket.blob(secret_name)
+            return blob.download_as_string()
+        except Exception as e:
+            logger.error(f"Failed to get secret {secret_name} from Google Cloud Storage: {e}")
+            return ""
+
+    def store_secret(self, secret_name: str, secret_value: str):
+        try:
+            blob = self.bucket.blob(secret_name)
+            blob.upload_from_string(secret_value)
+        except Exception as e:
+            logger.error(f"Failed to store secret {secret_name} in Google Cloud Storage: {e}")
+            raise e

pr_agent/secret_providers/secret_provider.py

@@ -0,0 +1,12 @@
+from abc import ABC, abstractmethod
+
+
+class SecretProvider(ABC):
+
+    @abstractmethod
+    def get_secret(self, secret_name: str) -> str:
+        pass
+
+    @abstractmethod
+    def store_secret(self, secret_name: str, secret_value: str):
+        pass