Add support for s3 tls-ca-chain

lib/charms/mysql/v0/s3_helpers.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 """S3 helper functions for the MySQL charms."""
-
+import base64
 import logging
 import tempfile
 import time
@@ -31,7 +31,7 @@
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 6
+LIBPATCH = 7
 
 # botocore/urllib3 clutter the logs when on debug
 logging.getLogger("botocore").setLevel(logging.WARNING)
@@ -52,13 +52,25 @@ def upload_content_to_s3(content: str, content_path: str, s3_parameters: Dict) -
     """
     try:
         logger.info(f"Uploading content to bucket={s3_parameters['bucket']}, path={content_path}")
+        ca_file = tempfile.NamedTemporaryFile()
         session = boto3.session.Session(
             aws_access_key_id=s3_parameters["access-key"],
             aws_secret_access_key=s3_parameters["secret-key"],
             region_name=s3_parameters["region"] or None,
         )
-
-        s3 = session.resource("s3", endpoint_url=s3_parameters["endpoint"])
+        verif = True
+        ca_chain = s3_parameters["tls-ca-chain"]
+        if ca_chain:
+            ca = "\n".join([base64.b64decode(s).decode() for s in ca_chain])
+            ca_file.write(ca.encode())
+            ca_file.flush()
+            verif = ca_file.name
+
+        s3 = session.resource(
+            "s3",
+            endpoint_url=s3_parameters["endpoint"],
+            verify=verif,
+        )
 
         bucket = s3.Bucket(s3_parameters["bucket"])
 
@@ -73,6 +85,8 @@ def upload_content_to_s3(content: str, content_path: str, s3_parameters: Dict) -
             exc_info=e,
         )
         return False
+    finally:
+        ca_file.close()
 
     return True
 

tests/unit/test_backups.py

@@ -54,13 +54,20 @@ def test_retrieve_s3_parameters(self, _get_s3_connection_info):
             "bucket": "test_bucket",
             "access-key": "test-access-key",
             "secret-key": "test-secret-key",
+            "tls-ca-chain": ["Zm9vYmFy"],  # "foobar" in base64
         }
         _get_s3_connection_info.return_value = return_value
 
         s3_parameters, missing_required_parameters = self.mysql_backups._retrieve_s3_parameters()
         self.assertEqual(
             s3_parameters,
-            {"endpoint": "https://s3.amazonaws.com", "region": None, "path": "", **return_value},
+            {
+                "endpoint": "https://s3.amazonaws.com",
+                "region": None,
+                "path": "",
+                "tls-ca-chain": "foobar",
+                **return_value,
+            },
         )
         self.assertEqual(missing_required_parameters, [])