Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: Store TLS related files in a writable storage #80

Merged
merged 1 commit into from
Oct 4, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
1 change: 1 addition & 0 deletions sekoia_automation/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from pathlib import Path

VOLUME_PATH = "/symphony"
TLS_VOLUME_PATH = "/tmp/tls"


def _json_load(value: str):
Expand Down
4 changes: 2 additions & 2 deletions sekoia_automation/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
12 changes: 12 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
16 changes: 8 additions & 8 deletions tests/test_storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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"),
}
Loading