Add support for s3 tls-ca-chain

Fixes #395

Signed-off-by: Peter Sabaini <peter.sabaini@canonical.com>

Author: sabaini

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
@@ -57,8 +57,20 @@ def upload_content_to_s3(content: str, content_path: str, s3_parameters: Dict) -
             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 = tempfile.NamedTemporaryFile()
+            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"])
 

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, [])