feat: support GCS repository for snapshot operations

lib/charms/opensearch/v0/constants_charm.py

@@ -148,3 +148,7 @@
 PERFORMANCE_PROFILE = "profile"
 
 JWT_CONFIG_RELATION = "jwt-configuration"
+
+# GCS Service account JSON
+
+GCS_SERVICE_ACCOUNT_JSON = "/var/snap/opensearch/common/home/snap_daemon/gcs_service_account.json"

lib/charms/opensearch/v0/helper_charm.py

@@ -10,7 +10,9 @@
 from types import SimpleNamespace
 from typing import TYPE_CHECKING, Iterable, List, Union
 
-from charms.opensearch.v0.constants_charm import PeerRelationName
+from charms.opensearch.v0.constants_charm import (
+    PeerRelationName,
+)
 from charms.opensearch.v0.helper_enums import BaseStrEnum
 from charms.opensearch.v0.models import App, PeerClusterApp
 from charms.opensearch.v0.opensearch_exceptions import OpenSearchCmdError

lib/charms/opensearch/v0/helper_security.py

@@ -2,6 +2,7 @@
 # See LICENSE file for licensing details.
 
 """Helpers for security related operations, such as password generation etc."""
+import json
 import logging
 import math
 import os
@@ -23,6 +24,8 @@
 from charms.opensearch.v0.models import ObjectStorageConfig
 from charms.opensearch.v0.opensearch_exceptions import OpenSearchCmdError
 from cryptography import x509
+from google.api_core.exceptions import GoogleAPIError
+from google.cloud import storage
 
 # The unique Charmhub library identifier, never change it
 LIBID = "224ce9884b0d47b997357fec522f11c7"
@@ -670,43 +673,41 @@ def verify_s3_credentials(cfg: ObjectStorageConfig) -> bool:
 
     All errors are logged with full traceback here.
     """
-    s3_cfg = cfg.s3
-
     ca_tmp_path = None
     verify_param: str | bool = True
 
     # If we have a custom CA chain, write it to a temp file and pass it to boto3
-    if s3_cfg.tls_ca_chain:
+    if cfg.s3.tls_ca_chain:
         fd, ca_tmp_path = tempfile.mkstemp(prefix="opensearch-s3-ca-", suffix=".pem")
         with os.fdopen(fd, "w") as f:
-            f.write(s3_cfg.tls_ca_chain)
+            f.write(cfg.s3.tls_ca_chain)
         verify_param = ca_tmp_path
 
     try:
         session = boto3.session.Session(
-            aws_access_key_id=s3_cfg.credentials.access_key,
-            aws_secret_access_key=s3_cfg.credentials.secret_key,
-            aws_session_token=getattr(s3_cfg.credentials, "session_token", ""),
+            aws_access_key_id=cfg.s3.credentials.access_key,
+            aws_secret_access_key=cfg.s3.credentials.secret_key,
+            aws_session_token=getattr(cfg.s3.credentials, "session_token", ""),
         )
 
         logger.info(
             "Verifying S3 with endpoint=%r bucket=%r region=%r has_ca=%r verify=%r",
-            s3_cfg.endpoint,
-            s3_cfg.bucket,
-            s3_cfg.region,
-            bool(s3_cfg.tls_ca_chain),
+            cfg.s3.endpoint,
+            cfg.s3.bucket,
+            cfg.s3.region,
+            bool(cfg.s3.tls_ca_chain),
             verify_param,
         )
 
         s3_client = session.client(
             "s3",
-            endpoint_url=s3_cfg.endpoint,
-            region_name=s3_cfg.region,
+            endpoint_url=cfg.s3.endpoint,
+            region_name=cfg.s3.region,
             verify=verify_param,
         )
 
         # This will test both credentials and TLS/CA
-        s3_client.head_bucket(Bucket=s3_cfg.bucket)
+        s3_client.head_bucket(Bucket=cfg.s3.bucket)
 
         logger.info("S3 credential validation with boto3 succeeded.")
         return True
@@ -739,24 +740,22 @@ def verify_azure_credentials(cfg: ObjectStorageConfig) -> bool:
     Uses the storage-account, secret-key and container fields provided by
     azure-storage-integrator.
     """
-    az_cfg = cfg.azure
-
     # TODO move this to the pydantic model validation
-    if az_cfg.connection_protocol not in {"http", "https"}:
+    if cfg.azure.connection_protocol not in {"http", "https"}:
         logger.warning(
             "Azure Storage credential validation failed: unsupported connection protocol %s",
-            az_cfg.connection_protocol,
+            cfg.azure.connection_protocol,
         )
         return False
 
     try:
-        account_name = az_cfg.credentials.storage_account
-        account_key = az_cfg.credentials.secret_key
-        container_name = az_cfg.container
+        account_name = cfg.azure.credentials.storage_account
+        account_key = cfg.azure.credentials.secret_key
+        container_name = cfg.azure.container
 
         # If azure integrator ever sends a custom endpoint, we will use it.
         # Otherwise, we will use public Azure blob endpoint.
-        raw_endpoint = az_cfg.endpoint
+        raw_endpoint = cfg.azure.endpoint
         account_url = raw_endpoint.rsplit("/", 1)[0]
         account_url = account_url or f"https://{account_name}.blob.core.windows.net"
 
@@ -779,3 +778,65 @@ def verify_azure_credentials(cfg: ObjectStorageConfig) -> bool:
             exc_info=e,
         )
         return False
+
+
+def verify_gcs_credentials(object_storage_config: ObjectStorageConfig) -> bool:  # noqa: C901
+    """Validate GCS credentials using google-cloud-storage.
+
+    Args:
+        cfg: ObjectStorageConfig object
+
+    Returns:
+        True if we can access the configured bucket, False otherwise.
+
+    cfg.gcs.credentials.secret_key to contain a service-account JSON
+    (as a string), and cfg.gcs.bucket to contain the bucket name.
+    """
+    if not object_storage_config.gcs.credentials:
+        logger.error("GCS credential validation failed: missing credentials block.")
+        return False
+
+    service_account_json = object_storage_config.gcs.credentials.secret_key
+    bucket_name = object_storage_config.gcs.bucket
+
+    if not service_account_json:
+        logger.error("GCS credential validation failed: secret_key is empty.")
+        return False
+    if not bucket_name:
+        logger.error("GCS credential validation failed: bucket name is missing.")
+        return False
+
+    try:
+        service_account = json.loads(service_account_json)
+    except (TypeError, ValueError) as e:
+        logger.error(
+            "GCS credential validation failed: secret_key is not valid JSON: %s",
+            e,
+        )
+        return False
+
+    client_kwargs: dict = {}
+    if project_id := service_account.get("project_id"):
+        client_kwargs["project"] = project_id
+
+    try:
+        client = storage.Client.from_service_account_info(
+            service_account,
+            **client_kwargs,
+        )
+        # list_blobs will raise if credentials are wrong or bucket is not accessible.
+        blobs_iter = client.list_blobs(bucket_name, max_results=1)
+        # Fetch one page
+        _ = next(iter(blobs_iter), None)
+
+        logger.info("GCS credential validation succeeded.")
+        return True
+
+    except (ValueError, TypeError, KeyError) as e:
+        # Invalid/missing service account fields, invalid private_key format
+        logger.error("GCS credential validation failed: invalid credentials: %s", e, exc_info=True)
+        return False
+
+    except GoogleAPIError as e:
+        logger.error("GCS credential validation failed: %s", e, exc_info=True)
+        return False

lib/charms/opensearch/v0/models.py

@@ -2,6 +2,8 @@
 # See LICENSE file for licensing details.
 
 """Cluster-related data structures / model classes."""
+import base64
+import binascii
 import json
 import logging
 import re
@@ -494,13 +496,38 @@ def from_relation(cls, input_dict: Optional[Dict[str, Any]]):
 class GcsRelDataCredentials(Model):
     """Model class for credentials passed on the gcs relation."""
 
-    secret_key: str = Field(alias="secret-key", default=None)
+    secret_key: Optional[str] = Field(alias="secret-key", default=None)
 
     class Config:
         """Model config of this pydantic model."""
 
         allow_population_by_field_name = True
 
+    @validator("secret_key", pre=True)
+    def _normalize_secret_key(cls, values):  # noqa: N805
+        """Accept either raw JSON or base64-encoded JSON"""
+        if values is None:
+            return None
+
+        content = values.decode() if isinstance(values, (bytes, bytearray)) else str(values)
+        if not (content := content.strip()):
+            return None
+
+        # already JSON
+        if content.startswith("{") and content.endswith("}"):
+            # validate JSON shape
+            json.loads(content)
+            return content
+
+        # base64 (urlsafe)
+        try:
+            decoded_bytes = base64.b64decode(content, altchars=b"-_", validate=True)
+            decoded_text = decoded_bytes.decode("utf-8").strip()
+            json.loads(decoded_text)
+            return decoded_text
+        except (binascii.Error, ValueError, UnicodeDecodeError, json.JSONDecodeError) as e:
+            raise ValueError("secret-key is not valid JSON (raw or base64-encoded)") from e
+
 
 class GcsRelData(Model):
     """Model class for the GCS relation data.
@@ -511,7 +538,56 @@ class GcsRelData(Model):
     bucket: str = Field(default="")
     base_path: Optional[str] = Field(alias="path", default=None)
     storage_class: Optional[str] = Field(alias="storage-class", default=None)
-    credentials: GcsRelDataCredentials = Field(alias=GCS_CREDENTIALS, default=None)
+    credentials: GcsRelDataCredentials = Field(
+        alias=GCS_CREDENTIALS, default_factory=GcsRelDataCredentials
+    )
+
+    class Config:
+        """Model config of this pydantic model."""
+
+        allow_population_by_field_name = True
+
+    @root_validator
+    def validate_core_fields(cls, values):  # noqa: N805
+        """Validate the core fields of the gcs relation data."""
+        creds = values.get("credentials")
+        if not creds or not creds.secret_key:
+            raise ValueError("Missing fields: secret-key")
+
+        if not values.get("bucket"):
+            raise ValueError("Missing field: bucket")
+
+        # remove any duplicate, prefix or trailing "/" characters
+        if base_path := values.get("base_path"):
+            base_path = re.sub(r"/+", "/", base_path).strip().strip("/")
+        values["base_path"] = base_path or None
+
+        return values
+
+    @validator(GCS_CREDENTIALS, check_fields=False)
+    def ensure_secret_content(cls, conf: Dict[str, str] | GcsRelDataCredentials):  # noqa: N805):
+        """Ensure the secret content is set."""
+        if not conf:
+            return None
+
+        data = conf if isinstance(conf, dict) else conf.dict(by_alias=True, exclude_none=True)
+        for v in data.values():
+            if isinstance(v, str) and v.startswith("secret://"):
+                raise ValueError(f"The secret content must be passed, received {v} instead")
+        return conf
+
+    @classmethod
+    def from_relation(cls, input_dict: Optional[Dict[str, Any]]):
+        """Create a new instance of this class from a json/dict repr.
+
+        This method creates a nested GcsRelDataCredentials object from the input dict.
+        """
+        if not input_dict:
+            return None
+        creds = GcsRelDataCredentials(**input_dict)
+        merged = {**input_dict}
+        merged[GCS_CREDENTIALS] = creds.dict(by_alias=True, exclude_none=True)
+        return cls.parse_obj(merged)
 
 
 class ObjectStorageConfig(Model):
@@ -534,7 +610,7 @@ class PeerClusterRelDataCredentials(Model):
     admin_tls: Optional[Dict[str, Optional[str]]]
     s3: Optional[S3RelDataCredentials]
     azure: Optional[AzureRelDataCredentials]
-    gcs: Optional[GcsRelData]
+    gcs: Optional[GcsRelDataCredentials]
 
 
 class PeerClusterApp(Model):

lib/charms/opensearch/v0/opensearch_peer_clusters.py

@@ -741,6 +741,12 @@ def rel_data_from_str_and_peek_secrets(self, redacted_dict_str: str) -> PeerClus
                 .peek_content()
                 .get("azure-secret-key")
             )
+        if credentials.get("gcs", {}).get("secret-key"):
+            credentials["gcs"]["secret-key"] = (
+                self._charm.model.get_secret(id=credentials["gcs"]["secret-key"])
+                .peek_content()
+                .get("gcs-secret-key")
+            )
         return PeerClusterRelData.from_dict(content)
 
     def _resolve_credential(
@@ -820,6 +826,11 @@ def rel_data_from_str(self, redacted_dict_str: str) -> PeerClusterRelData:
                 credentials["azure"]["secret-key"], content_key="azure-secret-key"
             )
 
+        if credentials.get("gcs", {}).get("secret-key"):
+            credentials["gcs"]["secret-key"] = self._resolve_credential(
+                credentials["gcs"]["secret-key"], content_key="gcs-secret-key"
+            )
+
         return PeerClusterRelData.from_dict(content)
 
     def is_any_cm_node_up_in_cluster(self) -> bool: