Merge pull request #80 from SEKOIA-IO/fix/tls_files_writable_storage

fix: Store TLS related files in a writable storage

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.5.2] - 2023-10-04
+
+### Fixed
+
+- Store TLS related files in a writable storage
+
 ## [1.5.1] - 2023-10-04
 
 ### Changed

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry]
 name = "sekoia-automation-sdk"
 
-version = "1.5.1"
+version = "1.5.2"
 description = "SDK to create Sekoia.io playbook modules"
 license = "MIT"
 readme = "README.md"

sekoia_automation/config.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 
 VOLUME_PATH = "/symphony"
+TLS_VOLUME_PATH = "/tmp/tls"
 
 
 def _json_load(value: str):

sekoia_automation/storage.py

@@ -10,7 +10,7 @@
 from tenacity import retry, stop_after_attempt, wait_exponential
 
 from sekoia_automation import constants
-from sekoia_automation.config import VOLUME_PATH, load_config
+from sekoia_automation.config import TLS_VOLUME_PATH, load_config
 from sekoia_automation.utils import capture_retry_error
 
 FilePath = Path | str
@@ -70,7 +70,7 @@ def _get_tls_client_credentials() -> tuple[Path | None, Path | None, Path | None
         * client.key
         * ca.crt
     """
-    volume = Path(VOLUME_PATH)
+    volume = Path(TLS_VOLUME_PATH)
     volume.mkdir(parents=True, exist_ok=True)
 
     ca_path = None

tests/conftest.py

@@ -62,6 +62,18 @@ def config_storage():
     config.VOLUME_PATH = old_config_storage
 
 
+@pytest.fixture
+def tls_storage():
+    old_tls_storage = config.TLS_VOLUME_PATH
+    config.TLS_VOLUME_PATH = mkdtemp()
+    storage_module.TLS_VOLUME_PATH = config.TLS_VOLUME_PATH
+
+    yield Path(config.TLS_VOLUME_PATH)
+
+    rmtree(config.TLS_VOLUME_PATH)
+    config.TLS_VOLUME_PATH = old_tls_storage
+
+
 @pytest.fixture
 def mocked_trigger_logs():
     with patch.object(

tests/test_storage.py

@@ -136,21 +136,21 @@ def test_get_tls_client_credentials_not_set():
     assert key is None
 
 
-def test_get_tls_client_credentials(config_storage):
+def test_get_tls_client_credentials(tls_storage):
     mocked = dict(CA_CERT="foo", CLIENT_CERT="bar", CLIENT_KEY="baz")
     with mock.patch.dict(os.environ, mocked):
         ca, cert, key = _get_tls_client_credentials()
-        assert ca == Path(config_storage).joinpath("ca.crt")
+        assert ca == Path(tls_storage).joinpath("ca.crt")
         assert Path(ca).exists()
 
-        assert cert == Path(config_storage).joinpath("client.crt")
+        assert cert == Path(tls_storage).joinpath("client.crt")
         assert Path(cert).exists()
 
-        assert key == Path(config_storage).joinpath("client.key")
+        assert key == Path(tls_storage).joinpath("client.key")
         assert Path(key).exists()
 
 
-def test_get_s3_data_path(config_storage):
+def test_get_s3_data_path(tls_storage):
     mocked = dict(
         AWS_BUCKET_NAME="bucket",
         AWS_ACCESS_KEY_ID="access_key",
@@ -169,10 +169,10 @@ def test_get_s3_data_path(config_storage):
         config: Config | None = first_call.kwargs.pop("config", None)
         assert config is not None
         assert config.client_cert == (
-            Path(config_storage).joinpath("client.crt"),
-            Path(config_storage).joinpath("client.key"),
+            Path(tls_storage).joinpath("client.crt"),
+            Path(tls_storage).joinpath("client.key"),
         )
         assert first_call.kwargs == {
             "endpoint_url": "https://aws-fake_url.com",
-            "verify": Path(config_storage).joinpath("ca.crt"),
+            "verify": Path(tls_storage).joinpath("ca.crt"),
         }